Python Programming Guide
Toolchain
- Type Checking:
ty check <FILE_PATH> --output-format concise - Lint:
ruff check --fix --unsafe-fixes <FILE_PATH> - Format:
ruff format <FILE_PATH> - Run tests:
uv run pytest - Sync deps:
uv sync --upgrade - NEVER use
python; useuv runinstead
References
When deeply modifying code in these areas, you MUST read the corresponding document first:
references/async-programming.md- before modifying anyio/async codereferences/type-hints-cheat-sheet.md- before modifying complex type annotationsreferences/exceptions.md- before modifying error handlingreferences/logging.md- before modifying logging
Code Standards
⚠️ You MUST fully understand and follow ALL rules demonstrated in the code example below BEFORE writing any code. Every RULE and ANTI-PATTERN shown is mandatory.
from __future__ import annotations
# RULE: Use relative imports for intra-package references, strictly limited to 2 levels (. or ..).
# RULE: Use absolute imports for references requiring 3 or more levels of nesting (...) to maintain readability.
from . import internal_module
from .. import parent_helper
from anyio import fail_after
# ANTI-PATTERN: from ...deep_parent import thing # (Too many dots!)
# ANTI-PATTERN: from mypackage.module import thing # (Use relative import for internal stuff)
import time
from collections.abc import Awaitable, Callable
from typing import Any, Protocol, TypedDict
import anyio
import attrs
from anyio import fail_after
from loguru import logger
class UserData(TypedDict):
"""
RULE: Use TypedDict for complex return values instead of plain dict/tuple.
ANTI-PATTERN: def get_user() -> dict[str, Any]: ...
ANTI-PATTERN: def get_coords() -> tuple[float, float, str]: ...
"""
user_id: str
name: str
email: str
# RULE: Use Python 3.12+ `type` statement for type aliases
# ANTI-PATTERN: JsonValue: TypeAlias = dict[str, Any] | list[Any] | str | int | float | bool | None
type JsonValue = dict[str, Any] | list[Any] | str | int | float | bool | None
class FetchResult(TypedDict):
"""RULE: Use TypedDict instead of tuple or dict for multi-value returns."""
data: UserData | None
error: str | None
class Validator(Protocol):
"""
RULE: Use Protocol with __call__ for complex callable signatures.
ANTI-PATTERN: Callable[[dict[str, Any], bool], bool]
"""
def __call__(self, data: dict[str, Any], *, strict: bool = False) -> bool: ...
@attrs.define
class Config:
"""
RULE: Use @attrs.define or @attrs.frozen for most classes, unless there is a specific need (e.g., Pydantic models, Exceptions, or TypedDict).
ANTI-PATTERN: class Foo: def __init__(self, x): self.x = x
"""
url: str
timeout: float = 10.0
headers: dict[str, str] = attrs.field(factory=dict)
retry: int = attrs.field(default=3)
@retry.validator
def _check_retry(self, attribute: attrs.Attribute[int], value: int) -> None:
"""RULE: Use attrs validators for field validation."""
if value < 0:
msg = "retry must be >= 0"
raise ValueError(msg)
@attrs.frozen
class DataSource:
"""
RULE: Use @attrs.frozen for immutable objects. Prefer composition over inheritance.
ANTI-PATTERN: class Base: def __getattr__(self, name): ...
ANTI-PATTERN: class Child(Parent(GrandParent)): ...
"""
name: str
endpoint: str
priority: int = 0
# --- Module: auth ---
class AuthError(Exception):
"""Base exception for the auth module."""
class InvalidTokenError(AuthError):
pass
# --- Module: service ---
class ServiceError(Exception):
"""
Base exception for the service module.
RULE: Define a base exception for each module to isolate error handling.
RULE: ServiceError and AuthError must be independent (no common base besides Exception).
"""
class NotFoundError(ServiceError):
"""RULE: Create specific exception types only when different handling is needed."""
def log_time[**P, R](func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
"""
RULE: Use Python 3.12+ type parameter syntax [T], [**P, R].
ANTI-PATTERN: T = TypeVar("T")
ANTI-PATTERN: P = ParamSpec("P")
ANTI-PATTERN: def func(x: T) -> T: ...
"""
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
start = time.perf_counter()
result = await func(*args, **kwargs)
print(f"{func.__name__}: {time.perf_counter() - start:.2f}s")
return result
return wrapper
@attrs.define
class ApiClient:
"""
RULE: Use anyio for ALL async code.
ANTI-PATTERN: asyncio.create_task(coro)
ANTI-PATTERN: await asyncio.gather(*coros)
"""
config: Config
sources: list[DataSource] = attrs.field(factory=list)
def __attrs_post_init__(self) -> None:
"""RULE: Use __attrs_post_init__ for complex initialization logic."""
self.sources.sort(key=lambda s: s.priority, reverse=True)
@log_time
async def fetch(
self,
user_id: str,
*,
timeout: float | None = None,
) -> UserData:
"""
RULE: Place parameters with defaults after * to force keyword-only usage.
RULE: No meaningless return values - raise or return None instead.
ANTI-PATTERN: return UserData(user_id="", name="", email="")
"""
timeout = timeout or self.config.timeout
errors: list[str] = []
for source in self.sources:
result = await self._try_fetch_from(source, user_id, timeout)
if result["data"]:
return result["data"]
if result["error"]:
errors.append(f"{source.name}: {result['error']}")
msg = f"User {user_id} not found in any source. Errors: {errors}"
raise NotFoundError(msg)
async def _try_fetch_from(
self, source: DataSource, user_id: str, timeout: float
) -> FetchResult:
"""
RULE: No pointless exception wrapping - let unexpected exceptions propagate.
ANTI-PATTERN: except Exception as e: raise FetchError(str(e)) from e
"""
try:
with fail_after(timeout):
data = await self._fetch_from(source, user_id)
return FetchResult(data=data, error=None)
except TimeoutError as e:
return FetchResult(data=None, error=f"timeout: {e}")
except ConnectionError as e:
return FetchResult(data=None, error=f"connection failed: {e}")
def _safe_shutdown(self) -> None:
"""
RULE: 'except Exception' is allowed ONLY if necessary (e.g., crash barriers).
RULE: MUST use '# noqa' AND provide an explanation.
ANTI-PATTERN:
try: ...
except Exception: pass
"""
try:
self.config = None
except Exception: # noqa: BLE001 to guarantees process exit
logger.exception("Cleanup failed")
async def _fetch_from(self, source: DataSource, user_id: str) -> UserData:
"""
RULE: Use concrete types instead of Any when structure is known.
ANTI-PATTERN: async def _fetch_from(self, source: Any, user_id: Any) -> Any: ...
"""
await anyio.sleep(0.1)
return UserData(
user_id=user_id, name=f"User {user_id}", email="test@example.com"
)
async def fetch_multiple(self, user_ids: list[str]) -> list[UserData]:
"""
RULE: Parallel tasks MUST use anyio.create_task_group().
RULE: Handle exceptions with except* (Python 3.11+) or ExceptionGroup.
"""
results: list[UserData] = []
failed: list[str] = []
async def fetch_and_collect(uid: str) -> None:
try:
data = await self.fetch(uid)
results.append(data)
except NotFoundError as e:
failed.append(f"{uid}: {e}")
try:
async with anyio.create_task_group() as tg:
for user_id in user_ids:
tg.start_soon(fetch_and_collect, user_id)
except* TimeoutError as eg:
for exc in eg.exceptions:
print(f"Timeout in parallel fetch: {exc}")
if failed:
print(f"Failed to fetch: {failed}")
return results
async def process(self, response: dict[str, Any] | list[Any] | str) -> str:
"""
RULE: Use match instead of chained if isinstance() checks.
RULE: Use static attribute access instead of getattr/setattr/hasattr.
ANTI-PATTERN: if isinstance(x, dict): ... elif isinstance(x, list): ...
ANTI-PATTERN: value = getattr(obj, "field")
"""
match response:
case {"status": "ok", "data": data}:
return f"Success: {data}"
case {"error": msg}:
return f"Error: {msg}"
case list() as items:
return f"Got {len(items)} items"
case str() as message:
return message
case _:
return "Unknown"
# ANTI-PATTERN: def create_wrong(**kwargs: Any) -> ApiClient: return ApiClient(config=Config(**kwargs))
async def create_correct(config: Config) -> ApiClient:
"""RULE: Accept constructed objects directly instead of **kwargs."""
return ApiClient(config=config)
# ANTI-PATTERN: config_path = Path(__file__).parent / "data" / "config.json"
def load_config_correct() -> dict[str, Any]:
"""RULE: Use importlib.resources.files() instead of __file__ path concatenation."""
import json
from importlib.resources import files
config_text = (files("mypackage") / "data" / "config.json").read_text()
return json.loads(config_text)
def process_data_shallow(items: list[str]) -> None:
"""
RULE: Keep indentation shallow (Linux kernel style). Max 3 levels.
RULE: Use guard clauses (early return/continue) to flatten logic.
ANTI-PATTERN:
if items:
for item in items:
if item.startswith("valid"):
process(item)
"""
if not items:
return
for item in items:
if not item.startswith("valid"):
continue
# process(item)
class SafeResource:
"""
RULE: APIs owning resources MUST be context managers.
RULE: Do NOT expose manual open/close methods.
ANTI-PATTERN:
res = Resource()
res.open() # Don't make users do this
res.close()
"""
def __enter__(self) -> SafeResource:
return self
def __exit__(self, *args: object) -> None:
self._cleanup()
def _cleanup(self) -> None:
pass
def use_builtin_resource(self) -> None:
"""
RULE: Never manually manage resources that support context managers.
ANTI-PATTERN:
f = open("file.txt")
try:
f.read()
finally:
f.close()
"""
with open("file.txt") as f:
_ = f.read()
@logger.catch(reraise=True)
async def main() -> None:
"""RULE: Use @logger.catch at entrypoints for automatic exception logging."""
config = Config(url="https://api.example.com", timeout=5.0)
sources = [
DataSource(name="primary", endpoint="https://api1.com", priority=10),
DataSource(name="backup", endpoint="https://api2.com", priority=5),
]
client = ApiClient(config=config, sources=sources)
user = await client.fetch("user123")
logger.info("Fetched user: {}", user["name"])
users = await client.fetch_multiple(["u1", "u2", "u3"])
logger.info("Fetched {} users", len(users))