| name | python-patterns |
| description | Pythonic idioms, PEP 8 standards, type hints, pytest testing strategies, TDD methodology, and best practices for building robust Python applications. |
Python Development Patterns
Idiomatic Python patterns and best practices for building robust, efficient, and maintainable applications.
When to Activate
- Writing new Python code
- Reviewing Python code
- Refactoring existing Python code
- Designing Python packages/modules
Core Principles
1. Readability Counts
Python prioritizes readability. Code should be obvious and easy to understand.
def get_active_users(users: list[User]) -> list[User]:
"""Return only active users from the provided list."""
return [user for user in users if user.is_active]
def get_active_users(u):
return [x for x in u if x.a]
2. Explicit is Better Than Implicit
Avoid magic; be clear about what your code does.
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
import some_module
some_module.setup()
3. EAFP - Easier to Ask Forgiveness Than Permission
Python prefers exception handling over checking conditions.
def get_value(dictionary: dict, key: str) -> Any:
try:
return dictionary[key]
except KeyError:
return default_value
def get_value(dictionary: dict, key: str) -> Any:
if key in dictionary:
return dictionary[key]
else:
return default_value
Type Hints
Basic Type Annotations
from typing import Optional, List, Dict, Any
def process_user(
user_id: str,
data: Dict[str, Any],
active: bool = True
) -> Optional[User]:
"""Process a user and return the updated User or None."""
if not active:
return None
return User(user_id, data)
Modern Type Hints (Python 3.9+)
def process_items(items: list[str]) -> dict[str, int]:
return {item: len(item) for item in items}
from typing import List, Dict
def process_items(items: List[str]) -> Dict[str, int]:
return {item: len(item) for item in items}
Type Aliases and TypeVar
from typing import TypeVar, Union
JSON = Union[dict[str, Any], list[Any], str, int, float, bool, None]
def parse_json(data: str) -> JSON:
return json.loads(data)
T = TypeVar('T')
def first(items: list[T]) -> T | None:
"""Return the first item or None if list is empty."""
return items[0] if items else None
Protocol-Based Duck Typing
from typing import Protocol
class Renderable(Protocol):
def render(self) -> str:
"""Render the object to a string."""
def render_all(items: list[Renderable]) -> str:
"""Render all items that implement the Renderable protocol."""
return "\n".join(item.render() for item in items)
Error Handling Patterns
Specific Exception Handling
def load_config(path: str) -> Config:
try:
with open(path) as f:
return Config.from_json(f.read())
except FileNotFoundError as e:
raise ConfigError(f"Config file not found: {path}") from e
except json.JSONDecodeError as e:
raise ConfigError(f"Invalid JSON in config: {path}") from e
def load_config(path: str) -> Config:
try:
with open(path) as f:
return Config.from_json(f.read())
except:
return None
Exception Chaining
def process_data(data: str) -> Result:
try:
parsed = json.loads(data)
except json.JSONDecodeError as e:
raise ValueError(f"Failed to parse data: {data}") from e
Custom Exception Hierarchy
class AppError(Exception):
"""Base exception for all application errors."""
pass
class ValidationError(AppError):
"""Raised when input validation fails."""
pass
class NotFoundError(AppError):
"""Raised when a requested resource is not found."""
pass
def get_user(user_id: str) -> User:
user = db.find_user(user_id)
if not user:
raise NotFoundError(f"User not found: {user_id}")
return user
Context Managers
Resource Management
def process_file(path: str) -> str:
with open(path, 'r') as f:
return f.read()
def process_file(path: str) -> str:
f = open(path, 'r')
try:
return f.read()
finally:
f.close()
Custom Context Managers
from contextlib import contextmanager
@contextmanager
def timer(name: str):
"""Context manager to time a block of code."""
start = time.perf_counter()
yield
elapsed = time.perf_counter() - start
print(f"{name} took {elapsed:.4f} seconds")
with timer("data processing"):
process_large_dataset()
Context Manager Classes
class DatabaseTransaction:
def __init__(self, connection):
self.connection = connection
def __enter__(self):
self.connection.begin_transaction()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.connection.commit()
else:
self.connection.rollback()
return False
with DatabaseTransaction(conn):
user = conn.create_user(user_data)
conn.create_profile(user.id, profile_data)
Comprehensions and Generators
List Comprehensions
names = [user.name for user in users if user.is_active]
names = []
for user in users:
if user.is_active:
names.append(user.name)
result = [x * 2 for x in items if x > 0 if x % 2 == 0]
def filter_and_transform(items: Iterable[int]) -> list[int]:
result = []
for x in items:
if x > 0 and x % 2 == 0:
result.append(x * 2)
return result
Generator Expressions
total = sum(x * x for x in range(1_000_000))
total = sum([x * x for x in range(1_000_000)])
Generator Functions
def read_large_file(path: str) -> Iterator[str]:
"""Read a large file line by line."""
with open(path) as f:
for line in f:
yield line.strip()
for line in read_large_file("huge.txt"):
process(line)
Data Classes and Named Tuples
Data Classes
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class User:
"""User entity with automatic __init__, __repr__, and __eq__."""
id: str
name: str
email: str
created_at: datetime = field(default_factory=datetime.now)
is_active: bool = True
user = User(
id="123",
name="Alice",
email="alice@example.com"
)
Data Classes with Validation
@dataclass
class User:
email: str
age: int
def __post_init__(self):
if "@" not in self.email:
raise ValueError(f"Invalid email: {self.email}")
if self.age < 0 or self.age > 150:
raise ValueError(f"Invalid age: {self.age}")
Named Tuples
from typing import NamedTuple
class Point(NamedTuple):
"""Immutable 2D point."""
x: float
y: float
def distance(self, other: 'Point') -> float:
return ((self.x - other.x) ** 2 + (self.y - other.y) ** 2) ** 0.5
p1 = Point(0, 0)
p2 = Point(3, 4)
print(p1.distance(p2))
Decorators
Function Decorators
import functools
import time
def timer(func: Callable) -> Callable:
"""Decorator to time function execution."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__} took {elapsed:.4f}s")
return result
return wrapper
@timer
def slow_function():
time.sleep(1)
Parameterized Decorators
def repeat(times: int):
"""Decorator to repeat a function multiple times."""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
results = []
for _ in range(times):
results.append(func(*args, **kwargs))
return results
return wrapper
return decorator
@repeat(times=3)
def greet(name: str) -> str:
return f"Hello, {name}!"
Class-Based Decorators
class CountCalls:
"""Decorator that counts how many times a function is called."""
def __init__(self, func: Callable):
functools.update_wrapper(self, func)
self.func = func
self.count = 0
def __call__(self, *args, **kwargs):
self.count += 1
print(f"{self.func.__name__} has been called {self.count} times")
return self.func(*args, **kwargs)
@CountCalls
def process():
pass
Concurrency Patterns
Threading for I/O-Bound Tasks
import concurrent.futures
import threading
def fetch_url(url: str) -> str:
"""Fetch a URL (I/O-bound operation)."""
import urllib.request
with urllib.request.urlopen(url) as response:
return response.read().decode()
def fetch_all_urls(urls: list[str]) -> dict[str, str]:
"""Fetch multiple URLs concurrently using threads."""
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_url = {executor.submit(fetch_url, url): url for url in urls}
results = {}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
results[url] = future.result()
except Exception as e:
results[url] = f"Error: {e}"
return results
Multiprocessing for CPU-Bound Tasks
def process_data(data: list[int]) -> int:
"""CPU-intensive computation."""
return sum(x ** 2 for x in data)
def process_all(datasets: list[list[int]]) -> list[int]:
"""Process multiple datasets using multiple processes."""
with concurrent.futures.ProcessPoolExecutor() as executor:
results = list(executor.map(process_data, datasets))
return results
Async/Await for Concurrent I/O
import asyncio
async def fetch_async(url: str) -> str:
"""Fetch a URL asynchronously."""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def fetch_all(urls: list[str]) -> dict[str, str]:
"""Fetch multiple URLs concurrently."""
tasks = [fetch_async(url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return dict(zip(urls, results))
Package Organization
Standard Project Layout
myproject/
├── src/
│ └── mypackage/
│ ├── __init__.py
│ ├── main.py
│ ├── api/
│ │ ├── __init__.py
│ │ └── routes.py
│ ├── models/
│ │ ├── __init__.py
│ │ └── user.py
│ └── utils/
│ ├── __init__.py
│ └── helpers.py
├── tests/
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_api.py
│ └── test_models.py
├── pyproject.toml
├── README.md
└── .gitignore
Import Conventions
import os
import sys
from pathlib import Path
import requests
from fastapi import FastAPI
from mypackage.models import User
from mypackage.utils import format_name
init.py for Package Exports
"""mypackage - A sample Python package."""
__version__ = "1.0.0"
from mypackage.models import User, Post
from mypackage.utils import format_name
__all__ = ["User", "Post", "format_name"]
Memory and Performance
Using slots for Memory Efficiency
class Point:
def __init__(self, x: float, y: float):
self.x = x
self.y = y
class Point:
__slots__ = ['x', 'y']
def __init__(self, x: float, y: float):
self.x = x
self.y = y
Generator for Large Data
def read_lines(path: str) -> list[str]:
with open(path) as f:
return [line.strip() for line in f]
def read_lines(path: str) -> Iterator[str]:
with open(path) as f:
for line in f:
yield line.strip()
Avoid String Concatenation in Loops
result = ""
for item in items:
result += str(item)
result = "".join(str(item) for item in items)
from io import StringIO
buffer = StringIO()
for item in items:
buffer.write(str(item))
result = buffer.getvalue()
Python Tooling Integration
Essential Commands
black .
isort .
ruff check .
pylint mypackage/
mypy .
pytest --cov=mypackage --cov-report=html
bandit -r .
pip-audit
safety check
pyproject.toml Configuration
[project]
name = "mypackage"
version = "1.0.0"
requires-python = ">=3.9"
dependencies = [
"requests>=2.31.0",
"pydantic>=2.0.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.4.0",
"pytest-cov>=4.1.0",
"black>=23.0.0",
"ruff>=0.1.0",
"mypy>=1.5.0",
]
[tool.black]
line-length = 88
target-version = ['py39']
[tool.ruff]
line-length = 88
select = ["E", "F", "I", "N", "W"]
[tool.mypy]
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--cov=mypackage --cov-report=term-missing"
Quick Reference: Python Idioms
| Idiom | Description |
|---|
| EAFP | Easier to Ask Forgiveness than Permission |
| Context managers | Use with for resource management |
| List comprehensions | For simple transformations |
| Generators | For lazy evaluation and large datasets |
| Type hints | Annotate function signatures |
| Dataclasses | For data containers with auto-generated methods |
__slots__ | For memory optimization |
| f-strings | For string formatting (Python 3.6+) |
pathlib.Path | For path operations (Python 3.4+) |
enumerate | For index-element pairs in loops |
Anti-Patterns to Avoid
def append_to(item, items=[]):
items.append(item)
return items
def append_to(item, items=None):
if items is None:
items = []
items.append(item)
return items
if type(obj) == list:
process(obj)
if isinstance(obj, list):
process(obj)
if value == None:
process()
if value is None:
process()
from os.path import *
from os.path import join, exists
try:
risky_operation()
except:
pass
try:
risky_operation()
except SpecificError as e:
logger.error(f"Operation failed: {e}")
Remember: Python code should be readable, explicit, and follow the principle of least surprise. When in doubt, prioritize clarity over cleverness.
Python Testing Patterns (pytest)
Comprehensive testing strategies for Python applications using pytest, TDD methodology, and best practices.
Test-Driven Development (TDD)
Always follow the TDD cycle:
- RED: Write a failing test for the desired behavior
- GREEN: Write minimal code to make the test pass
- REFACTOR: Improve code while keeping tests green
def test_add_numbers():
result = add(2, 3)
assert result == 5
def add(a, b):
return a + b
Coverage Requirements
- Target: 80%+ code coverage
- Critical paths: 100% coverage required
- Use
pytest --cov to measure coverage
pytest --cov=mypackage --cov-report=term-missing --cov-report=html
pytest Fundamentals
import pytest
def test_addition():
"""Test basic addition."""
assert 2 + 2 == 4
with pytest.raises(ValueError, match="invalid input"):
raise ValueError("invalid input provided")
Fixtures
@pytest.fixture
def database():
"""Fixture with setup and teardown."""
db = Database(":memory:")
db.create_tables()
yield db
db.close()
def test_database_query(database):
result = database.query("SELECT * FROM users")
assert len(result) > 0
Parametrization
@pytest.mark.parametrize("input,expected", [
("hello", "HELLO"),
("world", "WORLD"),
("PyThOn", "PYTHON"),
])
def test_uppercase(input, expected):
assert input.upper() == expected
Mocking
from unittest.mock import patch
@patch("mypackage.external_api_call")
def test_with_mock(api_call_mock):
api_call_mock.return_value = {"status": "success"}
result = my_function()
api_call_mock.assert_called_once()
assert result["status"] == "success"
Async Tests
@pytest.mark.asyncio
async def test_async_function():
result = await async_add(2, 3)
assert result == 5
Running Tests
pytest
pytest -v
pytest --cov=mypackage
pytest -m "not slow"
pytest -x
pytest --lf
pytest -k "test_user"
Best Practices
DO:
- Follow TDD: red-green-refactor
- Test one thing per test
- Use descriptive names:
test_user_login_with_invalid_credentials_fails
- Mock external dependencies
- Test edge cases: empty inputs, None values, boundaries
- Aim for 80%+ coverage on critical paths
DON'T:
- Test implementation details (test behavior)
- Share state between tests
- Ignore test failures
- Use
print in tests (use assertions)
- Write overly specific mocks that break easily