crefft Source 2026.10.3

Python Style Guide

crefft > Languages > Python

The key words “MUST”, “MUST NOT”, “REQUIRED”, “SHALL”, “SHALL NOT”, “SHOULD”, “SHOULD NOT”, “RECOMMENDED”, “MAY”, and “OPTIONAL” in this document are to be interpreted as described in RFC 2119.

Extends Google Python Style Guide.

Quick Reference

TaskToolCommand
LintRuff1uv run ruff check . --fix
FormatRuff1uv run ruff format .
Type checkMypy2uv run mypy src/
Type checkPyright3uv run pyright
SemanticSemgrep4uv run semgrep --config=p/python src/
Dead codeVulture5uv run vulture src/
Coveragepytest-cov6uv run pytest --cov=src
Complexity (report)Radon7uv run radon cc src/ -a
Complexity (gate)Ruff C9011uv run ruff check .
FuzzHypothesis8uv run pytest (with hypothesis tests)
Test perfpytest-xdist9uv run pytest -n auto
Password hashargon2-cffi10PasswordHasher().hash(password)
Encryptioncryptography11Fernet(key).encrypt(data)
Rate limitlimits12strategies.MovingWindowRateLimiter(storage)
Circuit breakpybreaker13@CircuitBreaker(fail_max=5)
Cache (memory)cachetools14@cached(TTLCache(maxsize=100, ttl=300))
Cache (disk)diskcache15Cache("/path").memoize()

Package Management: uv

Projects MUST use uv16 for all Python package management.

Why: uv manages project dependencies and environments together. Use uv sync --locked where the lockfile is authoritative.

# Create new project
uv init myproject
cd myproject

# Add dependencies
uv add requests pydantic

# Add dev dependencies
uv add --dev pytest ruff mypy

# Sync environment
uv sync

# Run commands in environment
uv run pytest
uv run ruff check .

pyproject.toml

All configuration MUST live in pyproject.toml. Projects MUST NOT use setup.py, setup.cfg, requirements.txt, or tool-specific config files.

Why: Centralizing all project configuration in pyproject.toml (PEP 518/621)17 eliminates configuration sprawl, reduces maintenance burden, and aligns with modern Python packaging standards.

Linting: Ruff

Projects MUST use Ruff1 for linting.

Why: Ruff replaces Flake8, isort, pyupgrade, and many other tools in a single, fast implementation. Written in Rust, it’s 10-100x faster than alternatives18 while providing comprehensive Python linting rules.

[tool.ruff]
target-version = "py314"
line-length = 100
src = ["src", "tests"]

[tool.ruff.lint]
select = [
    "E", "F", "W",      # pycodestyle, Pyflakes
    "I",                # isort
    "UP",               # pyupgrade
    "B",                # flake8-bugbear
    "C4",               # flake8-comprehensions
    "C90",              # mccabe (cyclomatic complexity, C901)
    "SIM",              # flake8-simplify
    "S",                # flake8-bandit (security)
    "N",                # pep8-naming
    "D",                # pydocstyle
    "ANN",              # flake8-annotations
    "RUF",              # Ruff-specific
]

[tool.ruff.lint.mccabe]
max-complexity = 10

[tool.ruff.lint.pydocstyle]
convention = "google"

Formatting: Ruff

Projects MUST use Ruff1 for code formatting.

Why: Ruff format is a drop-in replacement for Black19, but faster. Using the same tool for both linting and formatting simplifies tooling and improves performance.

[tool.ruff.format]
quote-style = "double"
indent-style = "space"
line-ending = "auto"

Type Checking: Mypy + Pyright

Projects MUST use both Mypy2 and Pyright3 for type checking.

Why: Mypy and Pyright catch different classes of type errors due to their different implementations and type inference strategies20. Run both checks; neither replaces the other in this standard.

Mypy (strict mode)

Projects MUST enable Mypy strict mode.

[tool.mypy]
strict = true
python_version = "3.14"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true

Pyright

Projects MUST enable Pyright strict mode.

[tool.pyright]
include = ["src"]
pythonVersion = "3.14"
typeCheckingMode = "strict"

Type Hint Conventions

Projects MUST follow these type hint conventions:

from collections.abc import Sequence, Mapping

# MUST use X | None, not Optional[X]
def get_user(id: int) -> User | None: ...

# MUST use collections.abc, not typing
def process(items: Sequence[str]) -> Mapping[str, int]: ...

Note: As of Python 3.14, PEP 649 deferred annotation evaluation is enabled by default. The from __future__ import annotations import is no longer required for forward references or avoiding import cycles in type hints.

PEP 695 Type Parameter Syntax (Python 3.12+)

Projects targeting Python 3.12+ SHOULD use the new type parameter syntax for generics and type aliases.

Why: PEP 695 provides a cleaner, more concise syntax for defining generic classes, functions, and type aliases. The new syntax eliminates boilerplate imports and makes type parameters more explicit and easier to read.

# Old style (pre-3.12)
from typing import TypeVar, Generic

T = TypeVar('T')
K = TypeVar('K')
V = TypeVar('V')

class Box(Generic[T]):
    def __init__(self, value: T) -> None:
        self.value = value

def first(items: list[T]) -> T:
    return items[0]

# New style (3.12+) - RECOMMENDED
class Box[T]:
    def __init__(self, value: T) -> None:
        self.value = value

def first[T](items: list[T]) -> T:
    return items[0]

def process_mapping[K, V](data: dict[K, V]) -> list[V]:
    return list(data.values())

# Type aliases with type parameters
type Point[T] = tuple[T, T]
type JSONDict = dict[str, str | int | float | bool | None]
type Result[T, E] = T | Exception

PEP 742 TypeIs for Type Narrowing (Python 3.13+)

Projects targeting Python 3.13+ SHOULD use TypeIs for user-defined type guards that narrow types.

Why: TypeIs provides more accurate type narrowing than TypeGuard. It correctly narrows the type in the else branch and works better with union types, improving type checker accuracy.

from typing import TypeIs

def is_string(val: object) -> TypeIs[str]:
    return isinstance(val, str)

def process(value: str | int) -> None:
    if is_string(value):
        # Type checker knows value is str here
        print(value.upper())
    else:
        # Type checker knows value is int here
        print(value + 1)

# More complex example
def is_non_empty_list[T](val: list[T] | None) -> TypeIs[list[T]]:
    return val is not None and len(val) > 0

PEP 750 Template Strings (Python 3.14+)

Projects targeting Python 3.14+ MAY use template strings (t-strings) for context-aware string interpolation.

Why: A t-string evaluates to a Template21 instead of a str, so the consumer decides how each interpolated value is escaped. An f-string has already been flattened by the time a function receives it; a Template still exposes the literal text and the interpolated values separately, which is what makes an escaping processor possible.

Projects MUST NOT treat a Template as inherently safe:

  • There is no is_safe() method.
  • str(template) returns the object’s repr, not the rendered text, so str(t"Hello {name}") is "Template(strings=('Hello ', ''), ...)".
  • Interpolated values are evaluated when the t-string is constructed. Only their processing is deferred, not their evaluation.

Safety comes entirely from the processor that consumes strings and interpolations.

from html import escape
from string.templatelib import Interpolation, Template

def render_html(template: Template) -> str:
    """Render a template, HTML-escaping every interpolated value."""
    parts: list[str] = []
    for item in template:
        if isinstance(item, Interpolation):
            parts.append(escape(str(item.value)))
        else:
            parts.append(item)
    return "".join(parts)

user_input = '<script>alert("xss")</script>'
render_html(t"<p>Hello {user_input}</p>")
# '<p>Hello &lt;script&gt;alert(&quot;xss&quot;)&lt;/script&gt;</p>'

SQL MUST stay parameterized. A t-string processor MUST emit a placeholder per interpolation and bind the values; it MUST NOT splice them into the statement text.

def to_sql(template: Template) -> tuple[str, list[object]]:
    """Turn a template into a parameterised statement and its bound values."""
    sql = template.strings[0]
    params: list[object] = []
    for interp, literal in zip(template.interpolations, template.strings[1:], strict=True):
        params.append(interp.value)
        sql += "?" + literal
    return sql, params

statement, params = to_sql(t"SELECT * FROM users WHERE name = {user_input}")
# ("SELECT * FROM users WHERE name = ?", ['<script>alert("xss")</script>'])
cursor.execute(statement, params)

Projects SHOULD prefer their rendering library’s own escaping (for example a template engine’s autoescaping, or the database driver’s parameter binding) over a hand-written processor, and MUST NOT reuse an HTML processor for SQL or vice versa: escaping is context-specific.

Standard Library Changes (Python 3.13+)

PEP 594 “Dead Batteries” Removal: Python 3.13 removed deprecated standard library modules. Projects MUST NOT rely on removed modules and SHOULD migrate to recommended alternatives:

  • aifc, audioop, chunk, cgi, cgitb, crypt, imghdr, mailcap, msilib, nis, nntplib, ossaudiodev, pipes, sndhdr, spwd, sunau, telnetlib, uu, xdrlib

New Modules: Python 3.14 introduces new standard library modules:

  • annotationlib: Low-level utilities for working with type annotations and introspection

Why: Removing unmaintained modules reduces the standard library maintenance burden and security surface. New modules like annotationlib provide better tools for runtime type introspection and metaprogramming.

Semantic Analysis: Semgrep

Projects SHOULD use Semgrep4 for semantic code analysis.

Why: Semgrep finds bugs, security issues, and anti-patterns using semantic code analysis22. It goes beyond syntax-level linting to understand code patterns and detect complex issues that traditional linters miss.

# Security audit
uv run semgrep --config=p/security-audit src/

# Python best practices
uv run semgrep --config=p/python src/

# OWASP Top 10
uv run semgrep --config=p/owasp-top-ten src/

Dead Code Detection: Vulture

Projects SHOULD use Vulture5 to detect dead code.

Why: Dead code increases maintenance burden, confuses developers, and can hide bugs. Vulture automatically identifies unused code with configurable confidence levels23.

uv run vulture src/ --min-confidence=80
[tool.vulture]
paths = ["src/"]
min_confidence = 80

Code Coverage: pytest-cov

Projects MUST measure code coverage and SHOULD maintain at least 80% coverage.

Why: Code coverage metrics help identify untested code paths24. Branch coverage ensures both sides of conditional logic are tested, catching bugs that line coverage alone would miss.

# Run with coverage
uv run pytest --cov=src --cov-branch --cov-report=html

# Fail if coverage drops below threshold
uv run pytest --cov=src --cov-fail-under=80
[tool.coverage.run]
source = ["src"]
branch = true

[tool.coverage.report]
fail_under = 80
exclude_lines = [
    "pragma: no cover",
    "if TYPE_CHECKING:",
    "@abstractmethod",
]

Cyclomatic Complexity: Ruff C901 + Radon

Projects MUST enforce a cyclomatic complexity ceiling of 10 with Ruff’s C901 rule1 and SHOULD use Radon7 for complexity reporting.

Why: High cyclomatic complexity indicates code that is difficult to test, understand, and maintain25. Enforcement and reporting are separate jobs: Ruff fails the build when a function exceeds the threshold, whereas Radon 6.0.1 only reports. Radon has no option to fail on a grade — radon cc src/ -a --fail B exits 2 with radon: error: unrecognized arguments: --fail B.

C901 is off unless C90 is selected, so the rule MUST appear in [tool.ruff.lint] select together with an explicit max-complexity. Installing Ruff is not enough on its own.

# Excerpt; see "Linting: Ruff" for the full select list.
[tool.ruff.lint]
select = ["C90"]

[tool.ruff.lint.mccabe]
max-complexity = 10
# Fail the build when any function exceeds complexity 10 (C901)
uv run ruff check .

# Report complexity grades (A=1-5, B=6-10, C=11-20, D=21-30, E=31-40, F=41+)
uv run radon cc src/ -a

Fuzzing: Hypothesis + Atheris

Projects SHOULD use property-based testing with Hypothesis8 and MAY use Atheris26 for coverage-guided fuzzing.

Hypothesis (Property-Based Testing)

Projects SHOULD use Hypothesis8 for property-based testing.

Why: Hypothesis is the world-class property-based testing library. It generates thousands of test cases automatically, finding edge cases that manual testing would miss27.

from hypothesis import given, strategies as st

@given(st.lists(st.integers()))
def test_sort_is_idempotent(xs: list[int]) -> None:
    assert sorted(sorted(xs)) == sorted(xs)

@given(st.text())
def test_encode_decode_roundtrip(s: str) -> None:
    assert s.encode().decode() == s

Atheris (Coverage-Guided Fuzzing)

Projects MAY use Atheris26 for coverage-guided fuzzing of native extensions or complex parsing code.

Why: Atheris uses coverage-guided fuzzing to find crashes and security issues in native extensions and complex parsing logic that property-based testing alone might miss28.

import atheris
import sys

def TestOneInput(data: bytes) -> None:
    try:
        my_parser(data.decode("utf-8", errors="ignore"))
    except ValueError:
        pass  # Expected

atheris.Setup(sys.argv, TestOneInput)
atheris.Fuzz()

Test Performance: pytest-xdist

Projects SHOULD use pytest-xdist9 to run tests in parallel.

Why: Running tests in parallel across CPU cores dramatically reduces test suite execution time29, enabling faster feedback loops and more frequent test runs.

# Auto-detect cores
uv run pytest -n auto

# Specific number of workers
uv run pytest -n 4

# Distribute by load
uv run pytest -n auto --dist=loadgroup
[tool.pytest.ini_options]
addopts = ["-n", "auto", "--dist", "loadgroup"]

Additional Performance Tips

TOML forbids a repeated key, so every pytest option MUST live in a single addopts array. A second [tool.pytest.ini_options] table fails to parse with TOMLDecodeError: Cannot declare ('tool', 'pytest', 'ini_options') twice, and a repeated addopts inside one table fails with Cannot overwrite a value — in both cases before pytest starts.

Flags MUST NOT be added unless the plugin defining them is installed: -n/--dist require pytest-xdist9, --reuse-db requires pytest-django30, and --testmon requires pytest-testmon31.

uv add --dev "pytest-xdist==3.8.0" "pytest-django==4.14.0" "pytest-testmon==2.2.0"
[tool.pytest.ini_options]
# One array only. Extend the array above rather than repeating the key.
addopts = ["-n", "auto", "--dist", "loadgroup", "--reuse-db", "--testmon"]
cache_dir = ".pytest_cache"

--reuse-db MUST NOT be used on a run that changes the schema, and --testmon MUST NOT be used for CI gating runs.

Why: Both flags trade completeness for speed. A reused database hides migration defects, and test selection skips tests whose dependencies changed in ways the plugin’s import graph did not observe.

Pre-commit Configuration

Hook revisions MUST be pinned to a tag that exists in the hook repository and MUST be new enough to parse the project’s target-version.

Why: Ruff 0.8.0 rejects this guide’s Ruff configuration outright — unknown variant py314, expected one of py37 ... py313 — and Mypy 1.13.0 cannot parse a t-string, reporting invalid syntax; you likely need to run mypy using Python 3.14 or newer. A stale pin is not a conservative choice; it is a lint stage that never runs.

repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.16.6
    hooks:
      - id: ruff-check
        args: [--fix]
      - id: ruff-format

  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v2.3.1
    hooks:
      - id: mypy
        args: [--strict]

  - repo: https://github.com/semgrep/semgrep
    rev: v1.176.0
    hooks:
      - id: semgrep
        args: [--config=p/python, --config=p/security-audit, --error]

  - repo: https://github.com/jendrikseipp/vulture
    rev: v2.16
    hooks:
      - id: vulture
        args: [src/, --min-confidence=90]

Migrating to These Pins

Read this before bumping an older configuration; each item changes behavior rather than just version numbers.

ChangeEffect
Semgrep moved to semgrep/semgrepreturntocorp/semgrep still redirects, but the canonical URL SHOULD be used
Ruff hook renamed to ruff-checkruff survives as a legacy alias, so renaming is RECOMMENDED, not required
Ruff 0.9 adopted the 2025 style guideruff format reformats existing code; land the reformat in its own commit
Ruff 0.16 expanded the default rule set from 59 to 413 rulesOnly affects projects with no explicit select; this guide sets one
Ruff 0.16 formats Python blocks inside MarkdownDocumentation files change on the first run
Mypy 2.0 enables --local-partial-typesCross-scope inference tightens; new errors are expected
Mypy 2.0 enables --strict-bytesbytearray and memoryview are no longer assignable to bytes (PEP 688)
Mypy 2.0 changed --allow-redefinitionBehaves as the old --allow-redefinition-new; --allow-redefinition-old restores the previous semantics

Why: Ruff’s formatter changes and Mypy’s default changes both produce diffs or errors that look like regressions in the pre-commit run that follows the bump. Landing the pin, the reformat, and the type fixes as separate commits keeps the cause of each diff legible.

CI Pipeline

# .github/workflows/ci.yml
jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - uses: astral-sh/setup-uv@1c37ad07a6a961277cf70c0d37d6f313000f5884 # v10.3.0
      - run: uv sync --locked
      - run: uv run ruff check .
      - run: uv run ruff format --check .
      - run: uv run mypy src/
      - run: uv run semgrep --config=p/python src/

  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - uses: astral-sh/setup-uv@1c37ad07a6a961277cf70c0d37d6f313000f5884 # v10.3.0
      - run: uv sync --locked
      - run: uv run pytest -n auto --cov=src --cov-fail-under=80

Dependencies & Package Management

Projects MUST:

  • Use uv16 for package management (uv sync, uv add, uv lock)
  • Commit lock files (uv.lock) with SHA256 hashes
  • Specify version constraints in pyproject.toml
  • Run uv sync --locked (or set UV_LOCKED=1) everywhere the lockfile is meant to be authoritative — CI, containers, and deployment
  • Scan for vulnerabilities using pip-audit32 or safety33
  • Configure Dependabot34 or Renovate35 for automated dependency updates

Why: Locked dependencies ensure reproducible builds36. Vulnerability scanning catches security issues early. Automated updates reduce maintenance burden while keeping dependencies current.

Lock enforcement MUST NOT be expressed as a [tool.uv] setting. There is no locked key: uv rejects it with unknown field locked, expected one of ... and then syncs anyway, writing a fresh uv.lock if none exists. Only the CLI flag or the environment variable actually enforce it.

# Lock dependencies
uv lock

# Install exactly what uv.lock pins; fail if it is missing or stale
uv sync --locked

# Update dependencies (rewrites uv.lock; commit the result)
uv sync --upgrade
[project]
dependencies = ["requests>=2.32.0,<3"]

A missing lockfile fails with error: Unable to find lockfile at 'uv.lock', but '--locked' was provided, and a lockfile that no longer matches pyproject.toml fails with error: The lockfile at 'uv.lock' needs to be updated, but '--locked' was provided. Both are the intended CI failures.

# Security scanning
uv run pip-audit
uv run safety check

E2E & Acceptance Testing

Projects with web interfaces SHOULD use Playwright37 for browser testing and MAY use pytest-bdd38 for BDD-style acceptance tests.

Why: Playwright provides reliable, cross-browser end-to-end testing with auto-waiting and built-in debugging39. pytest-bdd enables collaboration between technical and non-technical stakeholders using Gherkin syntax40.

from playwright.sync_api import Page, expect

def test_user_login(page: Page) -> None:
    page.goto("https://example.com/login")
    page.fill("#username", "test@example.com")
    page.click("button[type=submit]")
    expect(page.locator(".welcome")).to_be_visible()
# pytest-bdd example
from pytest_bdd import scenario, given, when, then

@scenario("features/login.feature", "User logs in successfully")
def test_login(): ...

@given("I am on the login page")
def on_login_page(page: Page) -> None:
    page.goto("/login")

Thread Safety Testing

Projects with concurrent code MUST test for thread safety and race conditions.

Why: Concurrent code is prone to race conditions and deadlocks that only manifest under specific timing conditions. Explicit thread safety testing catches these issues before production.

import threading
from concurrent.futures import ThreadPoolExecutor

class Counter:
    """Counter whose increments are serialized by a lock."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._value = 0

    def increment(self) -> None:
        with self._lock:
            self._value += 1

    @property
    def value(self) -> int:
        with self._lock:
            return self._value

def test_concurrent_access() -> None:
    counter = Counter()
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(counter.increment) for _ in range(100)]
        for future in futures:
            future.result()  # wait for completion and re-raise worker errors
    assert counter.value == 100

The test MUST await every future before asserting. Reading counter.value immediately after submit observes an arbitrary intermediate count — a correctly locked counter measured between 31 and 47 of 100 increments across 20 runs — so the test fails for reasons that have nothing to do with thread safety. The object under test MUST also be synchronized; without the lock above, the same test fails for the real reason.

# pytest-xdist distributes tests across worker processes; it has no
# --workers option and rejects it with "unrecognized arguments".
uv run pytest -n 4

-n 4 parallelizes the suite, not the threads inside a single test. It speeds up the run; the thread concurrency being verified still comes from the ThreadPoolExecutor in the test body.

Idempotence Testing

Projects SHOULD test that critical operations are idempotent.

Why: Idempotent operations can be safely retried without unintended side effects. This is essential for reliability in distributed systems, API design, and database operations where retries are common.

def test_create_user_idempotent(db: Session) -> None:
    user_id = create_user(db, email="test@example.com")
    same_id = create_user(db, email="test@example.com")
    assert user_id == same_id
    assert db.query(User).count() == 1
def test_payment_idempotent(api_client: Client) -> None:
    response1 = api_client.post("/pay", json={"amount": 100}, headers={"Idempotency-Key": "key1"})
    response2 = api_client.post("/pay", json={"amount": 100}, headers={"Idempotency-Key": "key1"})
    assert response1.json()["id"] == response2.json()["id"]

Reliability & Resilience Testing

Projects SHOULD test resilience patterns such as circuit breakers, retries, and fault tolerance.

Why: Testing resilience patterns ensures systems gracefully handle failures. Chaos engineering and fault injection reveal weaknesses before they cause production outages.

import pybreaker
import pytest

def test_circuit_breaker_opens(mocker) -> None:
    service = mocker.patch("external_service.call", side_effect=TimeoutError)
    circuit = pybreaker.CircuitBreaker(fail_max=5)

    # The first four failures propagate unchanged. The fifth reaches the
    # threshold, so pybreaker opens the circuit and raises CircuitBreakerError
    # instead of the upstream TimeoutError.
    for _ in range(4):
        with pytest.raises(TimeoutError):
            circuit.call(service)
    with pytest.raises(pybreaker.CircuitBreakerError):
        circuit.call(service)

    # `circuit.state` is a state object, not a string; compare current_state.
    assert circuit.current_state == pybreaker.STATE_OPEN
{
  "title": "Database failure",
  "method": [{
    "type": "action",
    "name": "terminate-db",
    "provider": {"type": "python", "module": "db.kill"}
  }]
}

Compatibility Testing

Projects SHOULD test across multiple Python versions and SHOULD test on multiple operating systems if targeting diverse environments.

Why: Testing across Python versions ensures compatibility as the ecosystem evolves. OS-specific testing catches platform-dependent bugs before users encounter them.

[tool.tox]
env_list = ["py312", "py313", "py314"]

# Native TOML: shared settings go under env_run_base, and each command is an
# array of arguments. A top-level [testenv] table is INI syntax; tox ignores it
# in pyproject.toml, leaving every environment with empty deps and commands.
[tool.tox.env_run_base]
deps = ["pytest==9.1.1", "pytest-cov==7.1.0"]
commands = [["pytest", "--cov=src"]]

Why: The INI-style form fails open. tox config reports commands = and deps = as empty for py312, py313, and py314, so the compatibility matrix passes on every Python version while running no tests at all. Verify with uv run tox config that each environment lists the expected commands before trusting a green matrix.

# GitHub Actions matrix
strategy:
  matrix:
    python-version: ["3.12", "3.13", "3.14"]
    os: [ubuntu-latest, macos-latest, windows-latest]

Note: Python 3.13+ includes experimental support for free-threading (PEP 703), removing the Global Interpreter Lock (GIL). Python 3.14 includes JIT compilation improvements for performance gains. Consider testing with free-threading enabled if your application has CPU-bound parallel workloads.

Internationalization Testing

Projects with international users SHOULD test UTF-8 handling and MAY test locale-specific behavior.

Why: Unicode handling bugs can corrupt data or cause crashes. Testing with diverse character sets ensures robust string handling across all languages and locales.

@pytest.mark.parametrize("text", ["Hello", "مرحبا", "你好", "🚀"])
def test_unicode_handling(text: str) -> None:
    result = process_text(text)
    assert result.encode("utf-8").decode("utf-8") == result
import babel.support

def test_translation() -> None:
    trans = babel.support.Translations.load("locale", ["es"])
    assert trans.gettext("Hello") == "Hola"

Data Integrity Testing

Projects with databases MUST test data integrity constraints and SHOULD test migration reversibility.

Why: Database constraints prevent data corruption. Testing migrations ensures schema changes can be safely deployed and rolled back. Backup/restore testing verifies disaster recovery procedures actually work.

def test_unique_constraint(db: Session) -> None:
    db.add(User(email="test@example.com"))
    db.commit()
    with pytest.raises(IntegrityError):
        db.add(User(email="test@example.com"))
        db.commit()
from pathlib import Path

import pytest
from alembic import command
from alembic.config import Config

@pytest.fixture
def alembic_config(tmp_path: Path) -> Config:
    """Alembic configuration pointed at a throwaway database."""
    config = Config("alembic.ini")
    config.set_main_option("script_location", "migrations")
    config.set_main_option("sqlalchemy.url", f"sqlite:///{tmp_path / 'test.db'}")
    return config

def test_migration_reversible(alembic_config: Config) -> None:
    command.upgrade(alembic_config, "head")
    command.downgrade(alembic_config, "base")
    command.upgrade(alembic_config, "head")  # Should succeed

Every Alembic command takes a Config as its first argument; there is no alembic.upgrade, and aliasing alembic.command as alembic does not supply the missing argument. downgrade(config, "base") drops every table the migrations created, so it MUST run against a database created for the test and MUST NOT run against a shared, staging, or production database.

A/B Testing & Feature Flags

Projects using feature flags MUST test both enabled and disabled states.

Why: Feature flags enable gradual rollouts and A/B testing, but untested flag states can cause production failures. Testing both states ensures features work correctly in all configurations.

@pytest.mark.parametrize("flag_enabled", [True, False])
def test_new_feature(flag_enabled: bool, mocker) -> None:
    mocker.patch("flags.is_enabled", return_value=flag_enabled)
    result = feature_under_test()
    assert result.uses_new_behavior == flag_enabled
import os

from flagsmith import Flagsmith

flagsmith = Flagsmith(environment_key=os.environ["FLAGSMITH_ENVIRONMENT_KEY"])

def checkout() -> Response:
    # Flags are evaluated on a Flags result, not on the client itself: the
    # client has no has_feature method.
    flags = flagsmith.get_environment_flags()
    if flags.is_feature_enabled("new_checkout"):
        return new_checkout_flow()
    return legacy_checkout_flow()

def checkout_for(user: User) -> Response:
    # Identity evaluation is required for segments and percentage rollouts.
    flags = flagsmith.get_identity_flags(identifier=user.id, traits={"plan": user.plan})
    if flags.is_feature_enabled("new_checkout"):
        return new_checkout_flow()
    return legacy_checkout_flow()

Tests MUST NOT reach the Flagsmith API. Use offline mode with a saved environment document so flag states are fixed and the suite stays hermetic:

from flagsmith import Flagsmith
from flagsmith.offline_handlers import LocalFileHandler

def test_checkout_uses_new_flow() -> None:
    client = Flagsmith(
        offline_mode=True,
        offline_handler=LocalFileHandler("tests/data/environment.json"),
    )
    assert client.get_environment_flags().is_feature_enabled("new_checkout")

Profiling

Projects SHOULD profile code before optimizing and MUST profile production workloads to identify actual bottlenecks.

Why: Premature optimization wastes time on non-critical paths. Profiling identifies where time is actually spent, enabling targeted optimization with measurable impact.

CPU Profiling

# Built-in cProfile
import cProfile
import pstats

with cProfile.Profile() as pr:
    expensive_function()

stats = pstats.Stats(pr)
stats.sort_stats("cumulative").print_stats(20)
# py-spy for production profiling (no code changes)
uv add --dev py-spy
py-spy top --pid 12345
py-spy record -o profile.svg --pid 12345  # Flame graph

Memory Profiling

from memory_profiler import profile

@profile
def memory_intensive() -> None:
    data = [i ** 2 for i in range(1000000)]
    return sum(data)
# Line-by-line memory usage
uv add --dev memory-profiler
python -m memory_profiler script.py

# Object tracking
uv add --dev objgraph
python -c "import objgraph; objgraph.show_most_common_types()"

Async Profiling

# aiomonitor for async introspection
uv add --dev aiomonitor
import asyncio

import aiomonitor

async def main() -> None:
    # start_monitor requires the running loop; calling it with no arguments
    # raises TypeError: missing 1 required positional argument: 'loop'.
    with aiomonitor.start_monitor(asyncio.get_running_loop()):
        await run_server()

The monitor binds 127.0.0.1 by default and offers an unauthenticated console. That default MUST be kept: exposing it on a routable interface hands out arbitrary introspection of a running process.

Continuous Profiling

For production, SHOULD use continuous profiling services:

# Pyroscope integration
import pyroscope

pyroscope.configure(
    application_name="myapp",
    server_address="http://pyroscope:4040",
)

Async Patterns

Projects using async/await MUST follow these patterns to avoid common pitfalls.

Event Loop Management

import asyncio

# MUST use asyncio.run() at top level
async def main() -> None:
    await do_work()

if __name__ == "__main__":
    asyncio.run(main())

# MUST NOT create event loops manually in library code
# BAD: loop = asyncio.get_event_loop()

Concurrent Execution

import asyncio

async def fetch_all(urls: list[str]) -> list[Response]:
    # Use gather for concurrent execution
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

# Use TaskGroup (Python 3.11+) for structured concurrency
async def fetch_all_safe(urls: list[str]) -> list[Response]:
    results: list[Response] = []
    async with asyncio.TaskGroup() as tg:
        for url in urls:
            tg.create_task(fetch_and_append(url, results))
    return results

Timeouts

import asyncio

# MUST use timeouts for external calls
async def fetch_with_timeout(url: str) -> Response:
    async with asyncio.timeout(30):  # Python 3.11+
        return await fetch(url)

# Or with wait_for
result = await asyncio.wait_for(fetch(url), timeout=30)

Resource Cleanup

# MUST use async context managers for cleanup
async with aiohttp.ClientSession() as session:
    response = await session.get(url)

# MUST NOT leave connections open
# BAD: session = aiohttp.ClientSession()  # Never closed

Blocking Code

import asyncio
from concurrent.futures import ThreadPoolExecutor

# MUST NOT block the event loop
# BAD: time.sleep(1)  # Blocks entire loop

# Use run_in_executor for blocking operations
async def run_blocking() -> None:
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_function)

Testing Async Code

import pytest

# Use pytest-asyncio
@pytest.mark.asyncio
async def test_fetch() -> None:
    result = await fetch("https://example.com")
    assert result.status == 200

# Or anyio for backend-agnostic testing
@pytest.mark.anyio
async def test_with_anyio() -> None:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(background_task())
[tool.pytest.ini_options]
asyncio_mode = "auto"

Security

Projects MUST follow secure coding practices for cryptography, secrets management, and password hashing.

Why: Security vulnerabilities in cryptographic code can expose user data, enable authentication bypass, and cause catastrophic breaches. Using well-audited libraries with secure defaults prevents common cryptographic mistakes.

Secrets Generation

Projects MUST use the secrets module (standard library) for generating cryptographically secure random values.

import secrets

# Generate secure random token
token = secrets.token_urlsafe(32)  # 256 bits of entropy

# Generate secure random bytes
random_bytes = secrets.token_bytes(32)

# Generate secure random hex string
hex_token = secrets.token_hex(32)

# Compare secrets in constant time to prevent timing attacks
if secrets.compare_digest(user_token, stored_token):
    authenticate_user()

Why: The random module is NOT cryptographically secure and MUST NOT be used for security-sensitive values like tokens, passwords, or API keys. The secrets module41 provides cryptographically strong random numbers suitable for security purposes.

Password Hashing

Projects MUST use argon2-cffi10 for password hashing.

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

ph = PasswordHasher(
    time_cost=3,        # Number of iterations
    memory_cost=65536,  # 64 MB memory
    parallelism=4,      # Parallel threads
)

# Hash password
hash = ph.hash("user_password")

# Verify password
try:
    ph.verify(hash, "user_password")
except VerifyMismatchError:
    raise AuthenticationError("Invalid password")

# Check if rehash needed (parameters changed)
if ph.check_needs_rehash(hash):
    new_hash = ph.hash("user_password")

Why: Argon2 won the Password Hashing Competition and provides memory-hard hashing that resists GPU and ASIC attacks. The argon2-cffi library wraps the official Argon2 C implementation with a safe Python API.

Cryptographic Operations

Projects SHOULD use the cryptography library11 for encryption, signing, and key management.

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import os

# Symmetric encryption with Fernet (AES-128-CBC + HMAC)
key = Fernet.generate_key()
fernet = Fernet(key)

# Encrypt
encrypted = fernet.encrypt(b"secret message")

# Decrypt
decrypted = fernet.decrypt(encrypted)

# Key derivation from password
salt = os.urandom(16)
kdf = PBKDF2HMAC(
    algorithm=hashes.SHA256(),
    length=32,
    salt=salt,
    iterations=600_000,  # OWASP 2023 recommendation
)
key = base64.urlsafe_b64encode(kdf.derive(b"password"))

Security Anti-Patterns

# BAD: Using random for security
import random
token = random.randint(0, 999999)  # Predictable!

# GOOD: Using secrets
import secrets
token = secrets.randbelow(1000000)

# BAD: Rolling your own crypto
import hashlib
hash = hashlib.sha256(password.encode()).hexdigest()  # No salt, too fast!

# GOOD: Using argon2
from argon2 import PasswordHasher
hash = PasswordHasher().hash(password)

# BAD: Hardcoding secrets
API_KEY = "sk-1234567890abcdef"  # Never commit secrets!

# GOOD: Environment variables
import os
API_KEY = os.environ["API_KEY"]

WebSocket Patterns

Projects using WebSockets SHOULD use the websockets library42 for async WebSocket communication.

Why: The websockets library is built on asyncio, provides a simple coroutine-based API, handles connection management automatically, and includes production-ready features like ping/pong, compression, and proper close handling.

Projects MUST use the websockets.asyncio implementation. The websockets.server, websockets.client, and websockets.legacy modules and the WebSocketServerProtocol type are the deprecated legacy implementation: importing them under websockets 17.1 emits websockets.server.serve is deprecated, websockets.legacy is deprecated, and websockets.WebSocketServerProtocol is deprecated. The two implementations MUST NOT be mixed — websockets.broadcast now resolves to websockets.asyncio.server.broadcast, which does not accept legacy protocol objects. The top-level websockets.serve and websockets.connect aliases already point at the asyncio implementation.

Server Implementation

import asyncio

from websockets.asyncio.server import ServerConnection, serve

async def handler(websocket: ServerConnection) -> None:
    async for message in websocket:
        # Echo back with processing
        await websocket.send(process_message(message))

async def main() -> None:
    async with serve(handler, "localhost", 8765):
        await asyncio.get_running_loop().create_future()  # Run forever

if __name__ == "__main__":
    asyncio.run(main())

Client Implementation

import asyncio

from websockets.asyncio.client import connect

async def client() -> None:
    uri = "ws://localhost:8765"
    async with connect(uri) as websocket:
        await websocket.send("Hello, Server!")
        response = await websocket.recv()
        print(f"Received: {response}")

asyncio.run(client())

Broadcast Pattern

from websockets.asyncio.server import ServerConnection, broadcast

CLIENTS: set[ServerConnection] = set()

async def handler(websocket: ServerConnection) -> None:
    CLIENTS.add(websocket)
    try:
        await websocket.wait_closed()
    finally:
        CLIENTS.discard(websocket)

def announce(message: str) -> None:
    """Queue a message on every open connection.

    broadcast() is synchronous and never awaits, so a slow client cannot
    block the sender.
    """
    broadcast(CLIENTS, message)

Connection Management

import asyncio

from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed

async def robust_client(uri: str) -> None:
    """Reconnecting WebSocket client with exponential backoff."""
    backoff = 1
    while True:
        try:
            async with connect(uri) as websocket:
                backoff = 1  # Reset on successful connection
                async for message in websocket:
                    await process_message(message)
        except (ConnectionClosed, OSError):
            # ConnectionClosed: the peer dropped an established connection.
            # OSError: the connection could not be opened at all.
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60)  # Max 60 seconds

Both failure modes MUST be handled. Catching only ConnectionClosed leaves the client dead as soon as the server is unavailable at connect time, because connect() raises OSError rather than ConnectionClosed in that case.

Rate Limiting

Projects making external API calls or exposing APIs SHOULD implement rate limiting using the limits library12 or pyrate-limiter43.

Why: Rate limiting prevents API abuse, protects against denial of service, ensures fair resource usage, and helps comply with upstream API provider limits.

Using limits Library

MovingWindowRateLimiter MUST be constructed with a storage instance. Calling it with no arguments raises TypeError: ... missing 1 required positional argument: 'storage'.

from limits import parse, storage, strategies

# In-memory storage (use Redis for distributed systems)
memory_storage = storage.MemoryStorage()
moving_window = strategies.MovingWindowRateLimiter(memory_storage)

# Define rate limit: 100 requests per minute
rate_limit = parse("100/minute")

def check_rate_limit(user_id: str) -> bool:
    """Check whether the request is allowed."""
    return moving_window.hit(rate_limit, user_id)

# With Redis storage for distributed systems
# redis_storage = storage.RedisStorage("redis://localhost:6379")

Token Bucket Implementation

import time
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    """Thread-safe token bucket rate limiter."""
    capacity: int
    refill_rate: float  # tokens per second
    tokens: float = field(init=False)
    last_refill: float = field(init=False)

    def __post_init__(self):
        self.tokens = float(self.capacity)
        self.last_refill = time.monotonic()

    def acquire(self, tokens: int = 1) -> bool:
        """Try to acquire tokens. Returns True if successful."""
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

# Usage: 10 requests per second with burst of 20
bucket = TokenBucket(capacity=20, refill_rate=10)
if bucket.acquire():
    make_api_call()
else:
    raise RateLimitExceeded()

Async Rate Limiting with pyrate-limiter

uv add "pyrate-limiter==4.5.0"
from pyrate_limiter import Duration, Limiter, Rate

# All rates MUST be passed as one list. Limiter's second positional parameter
# is buffer_ms, so Limiter(Rate(...), Rate(...)) silently binds the second rate
# to buffer_ms and enforces only the first.
limiter = Limiter([Rate(5, Duration.SECOND), Rate(100, Duration.MINUTE)])

async def rate_limited_call(user_id: str) -> dict:
    # PyrateLimiter 4.x has no ratelimit context manager. try_acquire_async
    # waits up to `timeout` seconds and returns whether capacity was granted.
    if not await limiter.try_acquire_async(user_id, blocking=True, timeout=5):
        raise RateLimitExceeded(user_id)
    return await make_api_call()

The boolean result MUST be checked before the call is made. blocking=True only bounds the wait; on timeout it returns False rather than raising, so an unchecked call proceeds straight past the limit.

Decorator Pattern

from functools import wraps
from limits import storage, strategies, parse

memory = storage.MemoryStorage()
limiter = strategies.FixedWindowRateLimiter(memory)

def rate_limit(limit_string: str):
    """Decorator for rate limiting functions."""
    rate = parse(limit_string)

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = func.__name__
            if not limiter.hit(rate, key):
                raise Exception("Rate limit exceeded")
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit("10/minute")
def send_email(to: str, subject: str):
    # Limited to 10 emails per minute
    ...

Circuit Breakers

Projects calling external services SHOULD implement circuit breakers using pybreaker13 to prevent cascading failures.

Why: Circuit breakers prevent your application from repeatedly calling a failing service, allowing it to recover and preventing resource exhaustion. They enable graceful degradation and faster failure detection.

Basic Usage

import pybreaker
import requests

# Create circuit breaker with 5 failure threshold
breaker = pybreaker.CircuitBreaker(
    fail_max=5,              # Open after 5 failures
    reset_timeout=60,        # Try again after 60 seconds
    exclude=[ValueError],    # Don't count these as failures
)

@breaker
def call_external_service(data: dict) -> dict:
    # (connect, read) timeouts in seconds. Without them requests waits
    # indefinitely and the breaker never sees a failure to count.
    response = requests.post(
        "https://api.example.com/v1/data",
        json=data,
        timeout=(3.05, 10),
    )
    response.raise_for_status()
    return response.json()

# Handle circuit breaker state
try:
    result = call_external_service({"key": "value"})
except pybreaker.CircuitBreakerError:
    # Circuit is open, use fallback
    result = get_cached_response()
except requests.RequestException:
    # Request failed but circuit may still be closed
    result = handle_failure()

Every production HTTP call MUST set an explicit timeout and MUST call raise_for_status() before decoding. The breaker counts whatever the wrapped function raises and exclude does not list: here requests.Timeout, requests.ConnectionError, and the requests.HTTPError produced by raise_for_status(). Without a timeout the call never raises, so the request occupies a worker indefinitely and the breaker cannot trip.

Note: timeout=(connect, read) bounds each socket operation, not the total wall-clock duration of the request. A response that streams slowly can still exceed the read timeout in aggregate; a hard deadline requires an outer cancellation mechanism.

With Listeners for Monitoring

import pybreaker
import logging

class LoggingListener(pybreaker.CircuitBreakerListener):
    def state_change(self, cb, old_state, new_state):
        logging.warning(
            f"Circuit {cb.name}: {old_state.name} -> {new_state.name}"
        )

    def failure(self, cb, exc):
        logging.error(f"Circuit {cb.name} failure: {exc}")

    def success(self, cb):
        logging.debug(f"Circuit {cb.name} success")

breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    listeners=[LoggingListener()],
    name="payment_service",
)

Async Circuit Breaker

For async code, use aiobreaker44:

uv add "aiobreaker==1.2.0"
from datetime import timedelta

import aiohttp
from aiobreaker import CircuitBreaker, CircuitBreakerError
from aiobreaker.state import CircuitBreakerState

# timeout_duration MUST be a timedelta. aiobreaker adds it to a datetime when
# the breaker opens, so an int raises
# TypeError: unsupported operand type(s) for +: 'datetime.datetime' and 'int'
# on the very failure path the breaker exists to handle.
breaker = CircuitBreaker(fail_max=5, timeout_duration=timedelta(seconds=60))

@breaker
async def async_external_call(
    session: aiohttp.ClientSession, url: str, data: dict
) -> dict:
    timeout = aiohttp.ClientTimeout(connect=3, total=10)
    async with session.post(url, json=data, timeout=timeout) as response:
        response.raise_for_status()
        return await response.json()

State transitions MUST be covered by a test, because the failure that opens the circuit is reported differently from the failures that precede it:

async def test_breaker_opens_after_threshold() -> None:
    breaker = CircuitBreaker(fail_max=2, timeout_duration=timedelta(seconds=60))

    @breaker
    async def failing() -> None:
        raise RuntimeError("upstream down")

    with pytest.raises(RuntimeError):
        await failing()
    with pytest.raises(CircuitBreakerError):  # threshold reached
        await failing()
    assert breaker.current_state is CircuitBreakerState.OPEN

Note: aiobreaker 1.2.0 was released in May 2021 and still calls the deprecated datetime.utcnow(), which warns on Python 3.12+. It works, and the release date alone does not prove abandonment, but projects SHOULD record it as a dependency risk and MUST pin it explicitly. pybreaker is not a substitute here: its call_async is a Tornado coroutine and raises NameError under plain asyncio.

Redis State Storage (Distributed)

import pybreaker
import redis

redis_client = redis.Redis(host="localhost", port=6379, db=0)

breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    state_storage=pybreaker.CircuitRedisStorage(
        pybreaker.STATE_CLOSED,
        redis_client,
        namespace="payment_breaker",
    ),
)

Caching

Projects SHOULD implement caching for expensive operations using cachetools14 for in-memory caching, diskcache15 for persistent caching, or aiocache45 for async caching.

Why: Caching reduces latency, decreases load on databases and external services, and improves application throughput. Choosing the right caching strategy depends on data freshness requirements, memory constraints, and whether caching needs to persist across restarts.

In-Memory Caching with cachetools

from cachetools import TTLCache, LRUCache, cached
from cachetools.keys import hashkey
import threading

# TTL cache: expire after 300 seconds, max 1000 items
cache = TTLCache(maxsize=1000, ttl=300)
lock = threading.Lock()  # cachetools is not thread-safe by default

@cached(cache, lock=lock)
def get_user(user_id: int) -> User:
    return db.query(User).filter_by(id=user_id).first()

# LRU cache with custom key function
@cached(LRUCache(maxsize=100), key=lambda x, y: hashkey(x.id, y))
def compute_score(user: User, category: str) -> float:
    return expensive_calculation(user, category)

# Manual cache operations
cache["custom_key"] = expensive_result
if "custom_key" in cache:
    result = cache["custom_key"]

Persistent Caching with diskcache

import diskcache

# File-backed cache (persists across restarts)
cache = diskcache.Cache("/tmp/my_cache")

# Store with TTL
cache.set("api_response", response_data, expire=3600)

# Memoization decorator
@cache.memoize(expire=600)
def fetch_data(url: str) -> dict:
    # Timeout first, then raise_for_status: without them a hung request blocks
    # the caller and an error body is cached for the full expiry window.
    response = requests.get(url, timeout=(3.05, 10))
    response.raise_for_status()
    return response.json()

# Atomic operations
with cache.transact():
    cache["key1"] = value1
    cache["key2"] = value2

# Cleanup
cache.expire()  # Remove expired items
cache.close()

Async Caching with aiocache

from aiocache import cached, Cache
from aiocache.serializers import JsonSerializer

# Simple async caching
@cached(ttl=300, cache=Cache.MEMORY)
async def get_user_async(user_id: int) -> dict:
    return await db.fetch_user(user_id)

# With Redis backend
@cached(
    ttl=300,
    cache=Cache.REDIS,
    serializer=JsonSerializer(),
    endpoint="localhost",
    port=6379,
)
async def get_external_data(key: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/{key}") as resp:
            return await resp.json()

Cache Invalidation Patterns

from cachetools import TTLCache

cache = TTLCache(maxsize=1000, ttl=300)

def invalidate_user_cache(user_id: int):
    """Invalidate specific cache entry."""
    cache.pop(f"user:{user_id}", None)

def invalidate_by_prefix(prefix: str):
    """Invalidate all entries matching prefix."""
    keys_to_delete = [k for k in cache if k.startswith(prefix)]
    for key in keys_to_delete:
        del cache[key]

# Event-driven invalidation
def on_user_updated(user_id: int):
    invalidate_user_cache(user_id)
    invalidate_by_prefix(f"user_scores:{user_id}")

Background Jobs

Projects SHOULD use concurrent.futures46 for simple parallelism and multiprocessing47 for CPU-bound work.

Why: Background jobs enable non-blocking execution of long-running tasks. Using standard library tools provides framework-agnostic patterns that work across all Python applications without external dependencies.

Thread Pool for I/O-Bound Work

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Callable

def process_urls(urls: list[str]) -> list[dict]:
    """Fetch multiple URLs in parallel."""
    results = []
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = {executor.submit(fetch_url, url): url for url in urls}
        for future in as_completed(futures):
            url = futures[future]
            try:
                result = future.result(timeout=30)
                results.append({"url": url, "data": result})
            except Exception as e:
                results.append({"url": url, "error": str(e)})
    return results

def run_with_timeout(fn: Callable, timeout: float, *args, **kwargs):
    """Run function with timeout."""
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(fn, *args, **kwargs)
        return future.result(timeout=timeout)

Process Pool for CPU-Bound Work

from concurrent.futures import ProcessPoolExecutor
import multiprocessing

def cpu_intensive_task(data: bytes) -> bytes:
    """CPU-bound processing (runs in separate process)."""
    return heavy_computation(data)

def process_batch(items: list[bytes]) -> list[bytes]:
    """Process items in parallel across CPU cores."""
    # Use number of CPUs minus 1 to leave headroom
    max_workers = max(1, multiprocessing.cpu_count() - 1)

    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(cpu_intensive_task, items))
    return results

Async Background Tasks

import asyncio
from typing import Coroutine, Any

class BackgroundTasks:
    """Manager for fire-and-forget async tasks."""

    def __init__(self):
        self._tasks: set[asyncio.Task] = set()

    def add(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task:
        """Add background task with automatic cleanup."""
        task = asyncio.create_task(coro)
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)
        return task

    async def shutdown(self, timeout: float = 30):
        """Wait for all tasks to complete."""
        if self._tasks:
            await asyncio.wait(self._tasks, timeout=timeout)

# Usage
background = BackgroundTasks()

async def handle_request(data: dict):
    # Fire and forget
    background.add(send_notification(data["user_id"]))
    return {"status": "accepted"}

async def shutdown():
    await background.shutdown()

Work Queue Pattern

import asyncio
from collections.abc import Callable, Coroutine
from typing import Any

class WorkQueue:
    """Async work queue with configurable concurrency."""

    def __init__(self, max_workers: int = 10):
        self._queue: asyncio.Queue = asyncio.Queue()
        self._workers: list[asyncio.Task] = []
        self._max_workers = max_workers

    async def start(self):
        """Start worker tasks."""
        for _ in range(self._max_workers):
            task = asyncio.create_task(self._worker())
            self._workers.append(task)

    async def _worker(self):
        while True:
            func, args, kwargs = await self._queue.get()
            try:
                await func(*args, **kwargs)
            except Exception as e:
                logging.error(f"Worker error: {e}")
            finally:
                self._queue.task_done()

    async def submit(
        self,
        func: Callable[..., Coroutine[Any, Any, Any]],
        *args,
        **kwargs,
    ):
        """Submit work to the queue."""
        await self._queue.put((func, args, kwargs))

    async def shutdown(self):
        """Wait for queue to drain and stop workers."""
        await self._queue.join()
        for worker in self._workers:
            worker.cancel()

Feature Flags

Projects SHOULD implement feature flags for gradual rollouts and A/B testing using a provider-agnostic pattern.

Why: Feature flags enable controlled rollouts, quick rollbacks, and A/B testing without deployment. A provider-agnostic interface allows switching between services (LaunchDarkly, PostHog, Flagsmith)48 without code changes.

Abstract Interface

from abc import ABC, abstractmethod
from typing import Any

class FeatureFlagProvider(ABC):
    """Abstract interface for feature flag providers."""

    @abstractmethod
    def is_enabled(
        self,
        flag_name: str,
        user_id: str | None = None,
        default: bool = False,
    ) -> bool:
        """Check if feature flag is enabled."""
        ...

    @abstractmethod
    def get_value(
        self,
        flag_name: str,
        user_id: str | None = None,
        default: Any = None,
    ) -> Any:
        """Get feature flag value (for multivariate flags)."""
        ...

Simple Implementation

import os
import json
from pathlib import Path

class EnvFeatureFlags(FeatureFlagProvider):
    """Environment variable-based feature flags."""

    def __init__(self, prefix: str = "FEATURE_"):
        self._prefix = prefix

    def is_enabled(
        self,
        flag_name: str,
        user_id: str | None = None,
        default: bool = False,
    ) -> bool:
        env_name = f"{self._prefix}{flag_name.upper()}"
        value = os.environ.get(env_name, "").lower()
        if value in ("1", "true", "yes", "on"):
            return True
        if value in ("0", "false", "no", "off"):
            return False
        return default

    def get_value(
        self,
        flag_name: str,
        user_id: str | None = None,
        default: Any = None,
    ) -> Any:
        env_name = f"{self._prefix}{flag_name.upper()}"
        value = os.environ.get(env_name)
        if value is None:
            return default
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            return value

# Usage
flags = EnvFeatureFlags()
if flags.is_enabled("new_checkout"):
    return new_checkout_flow()
else:
    return legacy_checkout_flow()

Percentage Rollout

import hashlib

class RolloutFeatureFlags(FeatureFlagProvider):
    """Feature flags with percentage-based rollout."""

    def __init__(self, config: dict[str, dict]):
        # Config: {"flag_name": {"enabled": True, "percentage": 50}}
        self._config = config

    def _get_bucket(self, flag_name: str, user_id: str) -> int:
        """Deterministic bucket assignment (0-99)."""
        key = f"{flag_name}:{user_id}"
        hash_value = hashlib.sha256(key.encode()).hexdigest()
        return int(hash_value[:8], 16) % 100

    def is_enabled(
        self,
        flag_name: str,
        user_id: str | None = None,
        default: bool = False,
    ) -> bool:
        config = self._config.get(flag_name, {})
        if not config.get("enabled", False):
            return False

        percentage = config.get("percentage", 100)
        if percentage >= 100:
            return True
        if percentage <= 0 or user_id is None:
            return False

        bucket = self._get_bucket(flag_name, user_id)
        return bucket < percentage

    def get_value(
        self,
        flag_name: str,
        user_id: str | None = None,
        default: Any = None,
    ) -> Any:
        config = self._config.get(flag_name, {})
        return config.get("value", default)

Testing with Feature Flags

import pytest
from unittest.mock import patch

@pytest.fixture
def feature_flags():
    """Fixture for testing with feature flags."""
    return {"new_checkout": True, "beta_feature": False}

@pytest.mark.parametrize("flag_enabled", [True, False])
def test_feature_behavior(flag_enabled: bool, mocker):
    mocker.patch.object(flags, "is_enabled", return_value=flag_enabled)
    result = feature_under_test()
    assert result.uses_new_behavior == flag_enabled

See Also

References

Footnotes

  1. Ruff Documentation ↩ ↩2 ↩3 ↩4 ↩5 ↩6

  2. Mypy Documentation ↩ ↩2

  3. Pyright Documentation ↩ ↩2

  4. Semgrep Documentation ↩ ↩2

  5. Vulture Documentation ↩ ↩2

  6. pytest-cov Documentation ↩

  7. Radon Documentation ↩ ↩2

  8. Hypothesis Documentation ↩ ↩2 ↩3

  9. pytest-xdist Documentation ↩ ↩2 ↩3

  10. argon2-cffi - Secure Password Hashing ↩ ↩2

  11. cryptography - Cryptographic Recipes ↩ ↩2

  12. limits - Rate Limiting Utilities ↩ ↩2

  13. pybreaker - Circuit Breaker Pattern ↩ ↩2

  14. cachetools - Extensible Memoizing Collections ↩ ↩2

  15. diskcache - Disk and File Backed Cache ↩ ↩2

  16. uv Documentation ↩ ↩2

  17. PEP 518 - pyproject.toml and PEP 621 - Project Metadata ↩

  18. Ruff Performance Benchmarks ↩

  19. Black - The Uncompromising Code Formatter ↩

  20. Mypy vs Pyright Comparison ↩

  21. string.templatelib - Template Strings and PEP 750 ↩

  22. Semgrep - Lightweight Static Analysis ↩

  23. Vulture - Find Dead Code ↩

  24. Code Coverage Best Practices ↩

  25. Cyclomatic Complexity - Wikipedia ↩

  26. Atheris - Coverage-Guided Fuzzer ↩ ↩2

  27. Hypothesis - Property-Based Testing ↩

  28. Atheris Documentation ↩

  29. pytest-xdist - Distributed Testing ↩

  30. pytest-django Documentation ↩

  31. pytest-testmon Documentation ↩

  32. pip-audit Documentation ↩

  33. Safety Documentation ↩

  34. Dependabot Documentation ↩

  35. Renovate Documentation ↩

  36. Reproducible Builds ↩

  37. Playwright for Python ↩

  38. pytest-bdd Documentation ↩

  39. Playwright - Reliable End-to-End Testing ↩

  40. Gherkin Syntax Reference ↩

  41. secrets - Generate Secure Random Numbers ↩

  42. websockets - WebSocket Library ↩

  43. pyrate-limiter - Leaky Bucket Rate Limiter ↩

  44. aiobreaker - Async Circuit Breaker ↩

  45. aiocache - Async Cache Library ↩

  46. concurrent.futures - Launching Parallel Tasks ↩

  47. multiprocessing - Process-Based Parallelism ↩

  48. Feature Flag Providers - LaunchDarkly, PostHog, Flagsmith ↩