python-project

name	python-project
description	Comprehensive guide for the entire Python programming process. This skill MUST be consulted before writing any code in a Python project to ensure adherence to toolchains, standards, and best practices.

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; use uv run instead

References

When deeply modifying code in these areas, you MUST read the corresponding document first:

• references/async-programming.md - before modifying anyio/async code
• references/type-hints-cheat-sheet.md - before modifying complex type annotations
• references/exceptions.md - before modifying error handling
• references/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="[e-mail]"
        )

    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))