loom-python

# Python Language Expertise ## Overview Idiomatic, production-grade Python: modern typing, async, packaging, and the web/data/ML ecosystem. Assumes competence — the value is in what is easy to get wrong (interning, mutable defaults, blocking the event loop, GIL, dataclass/Pydantic boundaries), not syntax tutorials. ## Tooling (uv / ruff / mypy) Ruff is a strong default for linting and formatting. `uv` is a capable project and environment manager, but preserve the repository's existing package manager and lockfile unless a migration is explicitly requested. **`uv`** — a single tool for project environments, locking, Python installation, and pip-compatible workflows. Use it for new work or repositories that already use it; do not treat it as an implicit replacement for Poetry or another established workflow. | Task | Command | | --- | --- | | New project | `uv init` (writes `pyproject.toml`) | | Add / remove dep | `uv add httpx` / `uv remove httpx` (updates `pyproject.toml` + `uv.lock`) | | Add dev dep | `uv add --dev pytest` | | Sync env from lock | `uv sync` (creates `.venv`, exact/reproducible) | | Run in project env | `uv run pytest` (auto-syncs first; no manual activate) | | Install a Python | `uv python install 3.12` | | Ephemeral tool | `uvx ruff check` (= `uv tool run`) | | pip shim | `uv pip install ...` (drop-in, no project file) | - In a uv-managed project, use `uv add`/`uv remove` so the manifest and `uv.lock` stay in sync; use `uv sync --frozen` in CI to fail on drift. **`ruff`** — single linter+formatter replacing flake8, isort, pyupgrade, pydocstyle, autoflake, and much of bandit. - `ruff check --fix` (lint+autofix), `ruff format` (near-identical to black, replaces it). - Config lives under `[tool.ruff.lint]` since 0.1 — top-level `select` is deprecated and warns. - Useful rule groups: `E`/`W` pycodestyle, `F` pyflakes, `I` isort, `UP` pyupgrade (keeps typing current), `B` bugbear (catches mutable defaults, `lru_cache` on methods), `SIM`, `S` bandit-security, `ASYNC`, `PL` pylint, `RUF`. Suppress with `# noqa: E501` (never bare `# noqa`). **`mypy --strict`** for type checking; `pyright`/`basedpyright` (what Pylance runs) is faster and stricter on inference. Always scope ignores: `# type: ignore[arg-type]`, never bare `# type: ignore`. ## Type Hints - 3.9+: built-in generics `list[T]`, `dict[K, V]`. 3.10+: `X | Y`, `X | None` (not `Union`/`Optional`). - Import `Callable`/`Sequence`/`Mapping`/`Iterator` from `collections.abc`, not `typing`. - **Accept abstract, return concrete:** take `Iterable`/`Sequence`/`Mapping` (max caller flexibility); return `list`/`dict` (callers know exactly what they get). - `Self` (3.11+) for fluent builders / alternate constructors. `TypedDict` for structured dicts (JSON payloads, `**kwargs`), `Protocol` for structural typing, `ParamSpec` for signature-preserving decorators — see Expert Practices. ```python from collections.abc import Iterable from typing import TypedDict, NotRequired, Self class MovieRow(TypedDict): # dict shape checkable by mypy title: str year: NotRequired[int] # 3.11+: optional key def deduplicate(items: Iterable[str]) -> list[str]: # abstract in, concrete out return list(dict.fromkeys(items)) class Builder: def with_name(self, name: str) -> Self: self._name = name return self ``` ## Async / Await Prefer `asyncio.TaskGroup` (3.11+) over `gather()`: on failure it cancels remaining sibling tasks, awaits them, and re-raises all errors as an `ExceptionGroup` (handled with `except*`). `gather()` does NOT cancel siblings on failure (they "won't be cancelled and will continue to run"), and `return_exceptions=True` silently mixes return values and exception objects into one list. Reserve `gather()` for genuinely independent fire-and-forget results. ⚠ **Never block the event loop.** A coroutine that calls blocking IO (`requests`, sync DB drivers, `open().read()`), `time.sleep()`, or CPU-heavy work freezes *every* task on that loop. Offload: `await asyncio.to_thread(fn, *args)` (3.9+) for blocking IO; `loop.run_in_executor(process_pool, fn)` for CPU work; `await asyncio.sleep()` not `time.sleep()`. See Async Gotchas for the full set. ```python import asyncio async def process_batch(urls: list[str]) -> list[dict]: async with asyncio.TaskGroup() as tg: tasks = [tg.create_task(fetch_data(url)) for url in urls] return [t.result() for t in tasks] # ExceptionGroup raised on any failure try: await process_batch(urls) except* ConnectionError as eg: # subgroup handling for exc in eg.exceptions: logger.error("fetch failed: %s", exc) rows = await asyncio.to_thread(cursor.execute, sql) # sync driver off the loop ``` ## Dataclasses vs Pydantic (the validation boundary) **Choose by trust, not habit.** | | `@dataclass` (stdlib) / `attrs` | Pydantic v2 `BaseModel` | | --- | --- | --- | | Validation | **None** — type hints are not enforced at runtime | Validates + coerces on construction | | Use for | Internal, already-trusted structs; hot paths | Untrusted edges: API bodies, config, external JSON | | Cost | Free | Per-instance validation (Rust core, but non-zero) | Don't trust a dataclass with external input (it will happily hold `age="banana"`); don't pay Pydantic's validation cost for internal structs constructed in a tight loop. For perf-critical validated decode, `msgspec` is the fastest option. ```python from dataclasses import dataclass, field from pydantic import BaseModel, Field, field_validator @dataclass(slots=True) # internal; slots=True saves memory + blocks typos class Config: host: str port: int = 8080 tags: list[str] = field(default_factory=list) # NEVER tags: list = [] class UserCreate(BaseModel): # untrusted edge email: str = Field(min_length=5) name: str = Field(max_length=100) @field_validator("email") @classmethod def norm_email(cls, v: str) -> str: if "@" not in v: raise ValueError("invalid email") # NEVER assert (-O strips it) return v.lower() ``` ⚠ Plain `@dataclass` sets `__hash__ = None` (unhashable — cannot be a set member / dict key). Use `frozen=True` for a hashable value object. `slots=True`/`kw_only=True` are 3.10+. ## Packaging (`pyproject.toml`) ```toml [project] name = "mypackage" version = "0.1.0" requires-python = ">=3.11" dependencies = ["httpx>=0.25", "pydantic>=2.0"] [dependency-groups] # PEP 735; `uv add --dev` targets this dev = ["pytest>=8", "pytest-asyncio>=0.23", "mypy>=1.9", "ruff>=0.4"] [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.ruff.lint] # MUST be nested since ruff 0.1; top-level is deprecated select = ["E4", "E7", "E9", "F", "I", "UP", "B", "SIM"] [tool.mypy] strict = true [tool.pytest.ini_options] asyncio_mode = "auto" # else async tests are silently UNCOLLECTED (default "strict") ``` ## Web Frameworks ### FastAPI Type hints drive validation, serialization, and OpenAPI. Use `Annotated[..., Depends(...)]` for injection; `response_model` to shape/filter output. ```python from typing import Annotated from fastapi import FastAPI, Depends, HTTPException, status app = FastAPI() async def get_db() -> AsyncIterator[AsyncSession]: # yield-dependency: cleanup after response async with AsyncSession(engine) as s: yield s @app.post("/users/", response_model=User, status_code=status.HTTP_201_CREATED) async def create_user(user: UserCreate, db: Annotated[AsyncSession, Depends(get_db)]) -> User: return await UserService(db).create(user) ``` ⚠ Gotchas: - **`def` vs `async def` path ops:** a `def` handler runs in FastAPI's threadpool (blocking libs OK). An `async def` handler runs on the event loop — any blocking call inside stalls *all* requests. Pick `def` for sync ORMs, `async def` only with async libs. - `response_model` strips fields not on the model — the primary guard against leaking hashes/tokens; don't return the raw ORM object. - `Depends` results are cached per-request; use `BackgroundTasks` for post-response work. ### Django ```python from django.db import models, transaction class User(models.Model): email = models.EmailField(unique=True) class Meta: indexes = [models.Index(fields=["email"])] with transaction.atomic(): # rolls back on exception user = User.objects.create(email=e) Profile.objects.create(user=user) ``` ⚠ QuerySets are lazy (evaluated on iteration/`len`/slice). Kill N+1 with `.select_related()` (FK/1:1 join) and `.prefetch_related()` (M2M/reverse). Batch with `bulk_create`/`bulk_update`. `.only()`/`.defer()` to trim columns. ## Data & ML (dense reference) The APIs are well-known; these are the correctness/perf traps that bite in review. **pandas** - **Vectorize.** `df["c"] = df["a"] * df["b"]`, `.map`/`np.where`/masks — never `iterrows`/`itertuples`/row-wise `apply` (Python-level loop, 10–100× slower). - **Assign via `.loc`**, never chained (`df[m]["c"] = ...`) — raises `SettingWithCopyWarning` and may no-op. pandas 3.0 makes Copy-on-Write the default, ending the ambiguity. - Memory: `read_csv(usecols=, dtype=)`; nullable `Int64`/`boolean` for NA-bearing ints; `category` dtype for low-cardinality strings; chunk with `read_csv(chunksize=)` or move to `polars`/`dask`/`pyarrow` at scale. - `merge` silently fans out on non-unique keys (row explosion) — validate with `merge(..., validate="one_to_many")`. **numpy** - Slices are **views**, not copies — `b = a[1:3]; b[0] = 9` mutates `a`. `.copy()` to break aliasing. - Broadcasting + ufuncs over Python loops; watch silent `int` overflow on fixed-width dtypes. - Use `rng = np.random.default_rng(seed)` (Generator API), not legacy global `np.random.seed`/`np.random.rand`. **scikit-learn** - Wrap preprocessing + estimator in a `Pipeline` so scalers/encoders fit on the **train fold only** inside CV — fitting a scaler on the full dataset before splitting leaks test statistics and inflates scores. `ColumnTransformer` for heterogeneous columns. Set `random_state` for reproducibility. **PyTorch** - Training step order: `optimizer.zero_grad()` → forward → `loss.backward()` → `optimizer.step()`. Forgetting `zero_grad` accumulates gradients. - Eval: `model.eval()` (disables dropout, freezes BatchNorm running stats) **and** `torch.inference_mode()` / `torch.no_grad()` (no autograd graph). Both are needed; they do different things. - Accumulate metrics as `loss.item()`/`.detach()` — keeping tensors alive retains the whole graph → memory blowup. `DataLoader(num_workers>0, pin_memory=True)` for GPU throughput; move batches with `.to(device, non_blocking=True)`. ## Testing with pytest Async fixtures must use `@pytest_asyncio.fixture` (a plain `@pytest.fixture` on `async def` yields an un-awaited coroutine under strict mode). Session/module-scoped async fixtures need `loop_scope=`; the `event_loop` fixture was removed in pytest-asyncio 1.0. Built-in fixtures worth reaching for: - `tmp_path` (per-test `pathlib.Path` dir), `monkeypatch` (`setattr`/`setenv`/`chdir`/`syspath_prepend`, all auto-undone at teardown — never patch globals by hand), `capsys`/`caplog`, `pytest.raises(ValueError, match="regex")`. - `conftest.py` fixtures are auto-discovered by every test in that directory tree — no import. Fixture `scope=` is `function` (default) / `class` / `module` / `session`; `autouse=True` applies without being requested. - Run tips: `-x` stop on first fail, `-k EXPR` select, `--lf` last-failed, `-q`, parametrize `ids=` for readable names. ```python import pytest, pytest_asyncio @pytest_asyncio.fixture async def db_session() -> AsyncIterator[AsyncSession]: async with AsyncSession(engine) as s: yield s @pytest.mark.parametrize("email,valid", [("a@b.com", True), ("bad", False)]) def test_email(email: str, valid: bool) -> None: if valid: User(email=email) else: with pytest.raises(ValueError): User(email=email) def test_env(monkeypatch): monkeypatch.setenv("API_KEY", "test") # auto-restored after the test assert load_config().key == "test" ``` ## Loom Test Runner Adapter