Python Rules

---
name: python-rules
description: "Team conventions for Python code (latest stable release only): docstrings and library documentation, modern typing syntax and mypy strict, exceptions, logging configured at the entry point, asyncio TaskGroup and cancellation, race-safe database writes, t-string and secret-handling security. Use when writing, fixing or reviewing Python code. Background knowledge that agents preload; not a command."
user-invocable: false
---

<!-- generated by knowledge.mjs from 39 note(s) (sha256 9ae1bdbc2ef8); edit the notes in knowledge/, then run build -->

# Python rules

Apply these rules to every Python file you write or review. The first section is always in force; open a rule from the table when the task touches it. Tooling, CI and the project layout are in `python-project-rules`.

## Always apply (critical, high)

### Never build a Template from a string that already contains user data

_critical · python_

**Rule.** Build SQL and HTML with a `t"..."` literal that keeps user values as interpolations, and pass the `Template` to a function that parameterizes or escapes each one. Never pass a string made with an f-string, `str.format` or `+` to `Template(...)`, and never render a `Template` to plain text and execute it.

**Why.** A string built before `Template(...)` is called makes the user's value part of the static template text, so the processing function can no longer tell data from code, which is plain SQL injection.

**Incorrect:**

```python
query = Template(f"SELECT * FROM users WHERE name = '{name}'")
cursor.execute(*to_sql(query))
```

**Correct:**

```python
query = t"SELECT * FROM users WHERE name = {name}"
cursor.execute(*to_sql(query))  # to_sql returns (sql, params)
```

### Put asyncio.timeout on every call to an external system

_high · python, asyncio_

**Rule.** Wrap each awaited call to a network, database or other external system in `async with asyncio.timeout(seconds)` and choose the value from what the caller can afford to wait.

**Why.** A call without a deadline waits as long as the other side does, so one stalled dependency holds the request, its connection and its worker indefinitely.

**Correct:**

```python
async def fetch(url):
    async with asyncio.timeout(5.0):
        return await client.get(url)
```

### Re-raise asyncio.CancelledError after cleanup

_high · python, asyncio_

**Rule.** When you catch `asyncio.CancelledError` to clean up, run the cleanup and then `raise` it again. Prefer `try`/`finally` when no other handling is needed.

**Why.** The asyncio docs warn that `asyncio.timeout()` and `TaskGroup` "might misbehave" when a coroutine swallows `CancelledError`: the task keeps running after it was told to stop.

**Correct:**

```python
async def worker():
    try:
        await job()
    except asyncio.CancelledError:
        await cleanup()
        raise
```

**Exceptions.** Suppress the cancellation only when this code deliberately owns it, and then call `Task.uncancel()` so the cancellation count stays correct.

### Prefer asyncio.TaskGroup over gather and loose create_task for concurrent work

_high · python, asyncio_

**Rule.** Start concurrent coroutines inside `async with asyncio.TaskGroup()` instead of `gather()` or unowned `create_task()`. Handle several failures with `except*`.

**Why.** A loose task outlives the code that started it and its error goes unobserved, while a `TaskGroup` cancels the remaining tasks on the first failure and raises an `ExceptionGroup`.

**Correct:**

```python
async def fetch_all(urls):
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(fetch(u)) for u in urls]
    return [task.result() for task in tasks]
```

**Exceptions.** Use `gather(..., return_exceptions=True)` when you want every result or error back and no cancellation of the others.

### Guard a check-then-write rule with one conditional UPDATE

_high · python, sqlalchemy_

**Rule.** When a rule reads before it writes (the last free copy), make the repository run one `UPDATE ... WHERE <rule>` and raise when it changed no row. The domain method keeps the readable rule; the UPDATE is what makes it hold under concurrent requests. Back a counter with a `CHECK` that states the rule (`CHECK (used <= max_uses)`), so a write that bypasses the UPDATE fails loudly.

**Why.** Two requests both read "1 copy free", both pass the check and both lend it, and a test that sends one request at a time never shows it.

**Correct:**

```python
result = cast("CursorResult[Any]", session.execute(   # the cast satisfies strict mypy: Result has no rowcount
    update(Book)
    .where(Book.id == book_id, Book.available_copies > 0)
    .values(available_copies=Book.available_copies - 1)
))
if result.rowcount == 0:
    raise NoFreeCopy(book_id)
```

**Exceptions.** A rule that counts rows (a per-member limit) cannot be one UPDATE: take the write lock before the read. On PostgreSQL lock the parent row with `SELECT ... FOR UPDATE`. On SQLite start the transaction with `BEGIN IMMEDIATE` through the SQLAlchemy `connect` and `begin` hooks, on an engine used for writes, because it serialises every transaction on that engine, and a second writer waits for the driver's `timeout` or fails with `SQLITE_BUSY`:

```python
@event.listens_for(engine, "connect")
def _manual_transactions(dbapi_connection, _record):
    dbapi_connection.isolation_level = None

@event.listens_for(engine, "begin")
def _begin_immediate(conn):
    conn.exec_driver_sql("BEGIN IMMEDIATE")
```

A process that never serves concurrent requests needs neither. A `UNIQUE` or `PRIMARY KEY` constraint already guards a plain uniqueness rule (once per customer); catch its error instead, as in `concurrency/python/map-only-the-unique-violation-and-roll-back`.

### Map only the UNIQUE violation to the domain error, roll back first and re-raise the rest

_high · python, sqlalchemy_

**Rule.** When a `UNIQUE` constraint guards a rule ("one seat per user"), catch the `IntegrityError`, call `session.rollback()`, and raise the domain error only when the violation is that constraint (PostgreSQL: `err.orig.diag.constraint_name`; SQLite: `err.orig.sqlite_errorcode` in `(sqlite3.SQLITE_CONSTRAINT_UNIQUE, sqlite3.SQLITE_CONSTRAINT_PRIMARYKEY)`, because a primary key reports its own code). Re-raise every other `IntegrityError`. Prove the rollback with a test that reuses the same session after the error and expects the next call to succeed.

**Why.** A `CHECK` or foreign-key failure reported as "already exists" hides a bug as a 409, a duplicate on a composite primary key that is not recognised as one turns into a 500, and a session left without `rollback()` after a failed flush raises `PendingRollbackError` on its next use.

**Correct:**

```python
try:                                   # Reservation has the primary key (event_id, user_id)
    session.add(Reservation(event_id=event_id, user_id=user_id))
    session.commit()
except IntegrityError as err:
    session.rollback()
    duplicate = (sqlite3.SQLITE_CONSTRAINT_UNIQUE, sqlite3.SQLITE_CONSTRAINT_PRIMARYKEY)
    if err.orig.sqlite_errorcode not in duplicate:   # PostgreSQL: diag.constraint_name
        raise
    raise AlreadyReserved(user_id) from err
```

**Exceptions.** A request-scoped session that is closed right after the error needs no explicit `rollback()`; keep it where a session is reused, as in a service that handles several commands on one session.

### Catch the specific exception, around only the statement that can raise it

_high · python_

**Rule.** Catch the exact exception you expect, around only the statement that can raise it. Never write a bare `except:`; use `except Exception:` for program errors.

**Why.** A bare `except:` also swallows `SystemExit` and `KeyboardInterrupt`, so the process cannot be stopped, and a wide `try` hides which line failed and turns a typo into a silent default.

**Correct:**

```python
def load_port(raw):
    config = parse(raw)
    try:
        return int(config["port"])
    except (KeyError, ValueError):
        return 8080
```

**Exceptions.** A bare `except:` is acceptable only when the handler logs and re-raises.

### Never leave a finally block with return, break or continue

_high · python_

**Rule.** Do not use `return`, `break` or `continue` to exit a `finally` block. Keep `finally` for cleanup, and return or compute the value outside it.

**Why.** Leaving `finally` that way silently discards an exception raised in the `try`, so a failure looks like success; Python 3.14 and newer report it as a `SyntaxWarning`, older versions accept it silently.

**Correct:**

```python
def read(path):
    handle = open(path)
    try:
        return handle.read()
    finally:
        handle.close()
```

### Log a caught exception with logger.exception, not with its text in the message

_high · python_

**Rule.** In the `except` block that finally handles the error, log with `logger.exception("Job %s failed", job.id)`. Do not use `logger.error(..., exc_info=True)` and do not append the exception to the message.

**Why.** A message such as `"failed: %s", err` records only the exception text, so the log says what failed but not where.

**Correct:**

```python
except JobError:
    logger.exception("Job %s failed", job.id)   # the layer that handles it
```

**Exceptions.** An expected domain error that a handler turns into a 4xx response (404, 409, 410) is not a failure of the service: log it with `logger.warning("Request refused: %s", exc)` and no traceback, so every unknown code is not an ERROR line with a stack. Keep `logger.exception` for the errors nobody expected; the catch-all `Exception` handler logs nothing here, see `api/python/fastapi/answer-unhandled-exceptions-with-a-generic-500-problem`.

### Never log secrets or whole payloads, and redact sensitive keys in a structlog processor

_high · python, structlog_

**Rule.** Do not log passwords, tokens, API keys, session ids, `Authorization` headers or whole request bodies; log selected fields. Add one processor before the renderer that masks sensitive keys, so no call site has to remember.

**Why.** A single `log.info("request", headers=headers)` writes bearer tokens into every log store and every backup of it.

**Correct:**

```python
SENSITIVE = {"password", "token", "authorization", "api_key", "secret"}

def redact(_, __, event_dict):
    for key in list(event_dict):
        if key.lower() in SENSITIVE:
            event_dict[key] = "***"
    return event_dict

structlog.configure(processors=[..., redact, renderer])
log.info("request", method=request.method, path=request.url.path)
```

**Exceptions.** Key-name redaction is defence in depth: a secret inside a value, a message or a traceback still leaks.

## Read when the task touches it

| Impact | Rule | Applies to |
| --- | --- | --- |
| medium | [Copy the values a response needs out of the session before you roll back](references/rules.md#copy-the-values-a-response-needs-out-of-the-session-before-you-roll-back) — After rollback() every loaded instance is expired, so the next attribute read starts a new transaction. | python, sqlalchemy |
| medium | [Build the documentation in CI with warnings treated as errors](references/rules.md#build-the-documentation-in-ci-with-warnings-treated-as-errors) — sphinx-build -W -n or mkdocs build --strict turns a broken reference into a failed build. | python |
| medium | [Declare the public API in __all__ of each package](references/rules.md#declare-the-public-api-in-__all__-of-each-package) — Without __all__ every name is public by accident, so renames break callers and API checks cannot tell what you promised. | python |
| medium | [Mark a public name with warnings.deprecated for at least one minor release](references/rules.md#mark-a-public-name-with-warningsdeprecated-for-at-least-one-minor-release) — Callers get a runtime warning and a type-checker hint first; the removal lands in a later release. | python |
| medium | [Run griffe check against the latest tag in CI](references/rules.md#run-griffe-check-against-the-latest-tag-in-ci) — griffe compares the public API with the last tag, so a removed parameter cannot ship in a minor release. | python, griffe |
| medium | [Generate the API reference from docstrings instead of retyping signatures](references/rules.md#generate-the-api-reference-from-docstrings-instead-of-retyping-signatures) — autodoc or mkdocstrings renders the signature from the code, so the reference cannot disagree with it. | python |
| medium | [Write docstring examples as doctests and run them in CI](references/rules.md#write-docstring-examples-as-doctests-and-run-them-in-ci) — An example nobody executes goes stale; pytest --doctest-modules turns it into a test. | python, pytest |
| medium | [Chain a replacement exception with raise X from err](references/rules.md#chain-a-replacement-exception-with-raise-x-from-err) — When one exception stands in for another, raise ... from err keeps the original traceback. | python |
| medium | [Log an error once, in the layer that finally handles it](references/rules.md#log-an-error-once-in-the-layer-that-finally-handles-it) — Intermediate layers add context and re-raise; only the handler at the boundary logs, so one failure is one entry. | any |
| medium | [Write logs to stdout or stderr and let the platform collect them](references/rules.md#write-logs-to-stdout-or-stderr-and-let-the-platform-collect-them) — A containerised app emits an event stream; rotation and shipping belong to the platform, not the app. | any |
| medium | [Configure logging once in the application entry point, never in a library](references/rules.md#configure-logging-once-in-the-application-entry-point-never-in-a-library) — A library only emits records (at most a NullHandler); the application sets handlers and format at startup. | python |
| medium | [Log the exception once at the boundary; layers in between only wrap it](references/rules.md#log-the-exception-once-at-the-boundary-layers-in-between-only-wrap-it) — Intermediate layers add context with raise ... from err; the HTTP handler, worker or main logs it. | python |
| medium | [Log through a module logger named with __name__, not print](references/rules.md#log-through-a-module-logger-named-with-__name__-not-print) — logger = logging.getLogger(__name__) in each module; diagnostic output goes through it, never through print(). | python |
| medium | [Name loggers with a constant, never with a runtime value](references/rules.md#name-loggers-with-a-constant-never-with-a-runtime-value) — Loggers are never released; put the varying value in the record instead. | python |
| medium | [Pass values to log calls as arguments, never through f-strings](references/rules.md#pass-values-to-log-calls-as-arguments-never-through-f-strings) — logger.info("Processed %s", order_id) formats only when the record is emitted and keeps one message template. | python |
| medium | [Bind request_id with structlog contextvars at the start of each request or task](references/rules.md#bind-request_id-with-structlog-contextvars-at-the-start-of-each-request-or-task) — Context variables are isolated per asyncio task; a module-level global is overwritten by concurrent requests. | python, structlog |
| medium | [Emit one JSON object per line in production and a readable console in development](references/rules.md#emit-one-json-object-per-line-in-production-and-a-readable-console-in-development) — Fixed fields (UTC timestamp, level, logger, event) parse in any aggregator; JSON also escapes control characters. | python, structlog |
| medium | [With structlog log a constant event name and put the data in snake_case fields](references/rules.md#with-structlog-log-a-constant-event-name-and-put-the-data-in-snake_case-fields) — log.info("order_processed", order_id=..., duration_ms=...) can be grouped and filtered; interpolated text cannot. | python, structlog |
| medium | [Run structlog on top of stdlib logging so library logs share one pipeline](references/rules.md#run-structlog-on-top-of-stdlib-logging-so-library-logs-share-one-pipeline) — Configure structlog with LoggerFactory and a ProcessorFormatter whose foreign_pre_chain matches its processors. | python, structlog |
| medium | [Point Python log handlers at sys.stdout, not at a file](references/rules.md#point-python-log-handlers-at-sysstdout-not-at-a-file) — StreamHandler on sys.stdout in the logging config; RotatingFileHandler only on hosts without a collector. | python |
| medium | [Make optional parameters of public functions keyword-only](references/rules.md#make-optional-parameters-of-public-functions-keyword-only) — Put optional parameters after a bare * so they can be added or reordered without breaking callers. | python |
| medium | [Run mypy in strict mode](references/rules.md#run-mypy-in-strict-mode) — strict = true makes an unannotated function an error, so the annotation rules are enforced. | python, mypy |
| medium | [Declare generics and type aliases with PEP 695 syntax, not TypeVar and Generic](references/rules.md#declare-generics-and-type-aliases-with-pep-695-syntax-not-typevar-and-generic) — class Box[T], def first[T] and type Maybe[T] = T \| None replace TypeVar, Generic and TypeAlias. | python |
| low | [Run an informational free-threaded Python job in CI before adopting that build](references/rules.md#run-an-informational-free-threaded-python-job-in-ci-before-adopting-that-build) — The free-threaded build is supported (PEP 779) but exposes races the GIL hid; run it as a non-blocking CI job first. | python |
| low | [Keep tutorials, how-to guides, reference and explanation on separate pages](references/rules.md#keep-tutorials-how-to-guides-reference-and-explanation-on-separate-pages) — The Diátaxis split: each page serves one reader need, so none mixes lessons, recipes, facts and background. | any |
| low | [Record every public change in CHANGELOG.md in the Keep a Changelog format](references/rules.md#record-every-public-change-in-changelogmd-in-the-keep-a-changelog-format) — Each pull request that changes public behaviour adds an entry under Unreleased, grouped by kind. | any |
| low | [Write docstrings for public API in Google style without repeating annotated types](references/rules.md#write-docstrings-for-public-api-in-google-style-without-repeating-annotated-types) — PEP 257 sets the shape; Google style (Args, Returns, Raises) gives sections, and the signature already states types. | python |
| low | [Import names explicitly through absolute paths, never with a star](references/rules.md#import-names-explicitly-through-absolute-paths-never-with-a-star) — Prefer absolute imports and never from module import *, so every name has a visible origin. | python |
| low | [Write optional types as X \| None, never Optional[X]](references/rules.md#write-optional-types-as-x--none-never-optionalx) — The PEP 604 union operator needs no import and reads the same in every annotation. | python |