Mastering Python Skill

Mastering Python Skill

Production-ready Python patterns with runnable code examples.

Contents

• Workflow
• Reference Files
• Sample CLI Tools
• When NOT to Use
• Full Table of Contents

---

Workflow

Phase 1: Setup

1. Verify Python version

   python --version  # Require 3.10+, prefer 3.12+

2. Create and activate virtual environment

   python -m venv .venv && source .venv/bin/activate

3. Install dependencies

   poetry install  # or: pip install -r requirements.txt

Phase 2: Develop

4. Reference appropriate patterns:

• Types → type-systems.md
• Async → async-programming.md
• APIs → fastapi-patterns.md
• DB → database-access.md

5. Follow project structure from project-structure.md

Phase 3: Validate

6. Run quality checks

   ruff check . && ruff format --check .
   mypy src/

7. Run tests with coverage

   pytest -v --cov=src --cov-report=term-missing

Phase 4: Deploy

8. Build and verify package

   python -m build && twine check dist/*

9. Deploy per docker-deployment.md or ci-cd-pipelines.md

Pre-Completion Checklist:

- [ ] All tests pass
- [ ] mypy reports no errors
- [ ] ruff check clean
- [ ] Coverage ≥80%
- [ ] No security warnings in dependencies

---

Reference Files

See TOC.md for detailed topic lookup.

---

Sample CLI Tools

Runnable examples demonstrating production patterns:

# Test examples
python sample-cli/async_fetcher.py https://httpbin.org/get
python sample-cli/config_loader.py --show-env
python sample-cli/db_cli.py init --sample-data && python sample-cli/db_cli.py list
python sample-cli/code_validator.py src/

---

When NOT to Use

• Non-Python languages: Use language-specific skills
• ML/AI model internals: Use PyTorch/TensorFlow skills
• Cloud infrastructure: Use AWS/GCP skills for infra (this covers code)
• Legacy Python 2: Focus is Python 3.10+

Code Quality, Linting, and Formatting

Contents

• Quick Snippets
• Core Concepts
• Production Examples
• Example 1: Ruff Configuration
• Example 2: Pre-commit Hooks
• Example 3: CI/CD Quality Gates
• Example 4: Measuring Code Quality
• Common Patterns
• Pitfalls to Avoid
• See Also

---

Quick Snippets

---

Core Concepts

Code quality tools automate enforcement of consistent standards:

• Linters (Ruff, Pylint) detect errors and anti-patterns
• Formatters (Ruff, Black) enforce consistent style
• Type checkers (mypy, pyright) catch type errors statically
• Complexity analyzers (radon) identify hard-to-maintain code

The modern Python stack:

• Ruff: 10-100x faster than traditional tools, replaces Flake8 + isort + Black
• mypy: Industry-standard type checker
• pre-commit: Git hooks for automated quality checks

---

Production Examples

Example 1: Ruff Configuration

Use case: Configure Ruff as an all-in-one linter and formatter.

# pyproject.toml - Ruff configuration

[tool.ruff]
# Target Python version
target-version = "py312"

# Line length (matches Black)
line-length = 88

# Source directories
src = ["src", "tests"]

# Exclude paths
exclude = [
    ".git",
    ".venv",
    "__pycache__",
    "build",
    "dist",
    ".eggs",
]

[tool.ruff.lint]
# Enable rule categories
select = [
    "E",      # pycodestyle errors
    "W",      # pycodestyle warnings
    "F",      # Pyflakes
    "I",      # isort
    "B",      # flake8-bugbear
    "C4",     # flake8-comprehensions
    "UP",     # pyupgrade
    "ARG",    # flake8-unused-arguments
    "SIM",    # flake8-simplify
    "TCH",    # flake8-type-checking
    "PTH",    # flake8-use-pathlib
    "ERA",    # eradicate (commented code)
    "PL",     # Pylint
    "RUF",    # Ruff-specific rules
]

# Ignore specific rules
ignore = [
    "E501",   # Line too long (handled by formatter)
    "PLR0913", # Too many arguments
]

# Allow autofix for all enabled rules
fixable = ["ALL"]
unfixable = []

# Per-file ignores
[tool.ruff.lint.per-file-ignores]
"tests/*" = [
    "ARG",    # Unused arguments in tests (fixtures)
    "PLR2004", # Magic values in tests
]
"__init__.py" = ["F401"]  # Unused imports in __init__

[tool.ruff.lint.isort]
known-first-party = ["myproject"]
force-single-line = false
lines-after-imports = 2

[tool.ruff.lint.pylint]
max-args = 7
max-branches = 12

[tool.ruff.format]
# Formatting options
quote-style = "double"
indent-style = "space"
skip-magic-trailing-comma = false
line-ending = "auto"

Running Ruff:

#!/bin/bash
# scripts/lint.sh - Linting script

set -e

echo "=== Running Ruff Linter ==="
ruff check .

echo "=== Running Ruff Formatter Check ==="
ruff format --check .

echo "=== All checks passed! ==="

Auto-fix and format:

# Fix linting issues and format in one command
ruff check --fix . && ruff format .

Key points:

• Ruff replaces Flake8, isort, and Black with one fast tool
• Use select to enable rule categories by prefix
• per-file-ignores handles special cases like tests
• Run in CI to enforce standards

---

Example 2: Pre-commit Hooks

Use case: Automatically run quality checks before every commit.

# .pre-commit-config.yaml

# Pre-commit configuration
# Run: pre-commit install (once)
# Run manually: pre-commit run --all-files

repos:
  # Ruff - Linting and formatting
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.1.9
    hooks:
      - id: ruff
        args: [--fix]
      - id: ruff-format

  # mypy - Type checking
  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.7.0
    hooks:
      - id: mypy
        additional_dependencies:
          - pydantic>=2.0
          - types-requests
        args: [--strict, --ignore-missing-imports]

  # General pre-commit hooks
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.5.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml
      - id: check-json
      - id: check-toml
      - id: check-added-large-files
        args: ['--maxkb=1000']
      - id: check-merge-conflict
      - id: detect-private-key
      - id: no-commit-to-branch
        args: ['--branch', 'main', '--branch', 'master']

  # Security checks
  - repo: https://github.com/PyCQA/bandit
    rev: 1.7.6
    hooks:
      - id: bandit
        args: ['-c', 'pyproject.toml']
        additional_dependencies: ['bandit[toml]']

  # Commit message format
  - repo: https://github.com/commitizen-tools/commitizen
    rev: v3.13.0
    hooks:
      - id: commitizen
        stages: [commit-msg]

# CI configuration
ci:
  autofix_commit_msg: |
    [pre-commit.ci] auto fixes from pre-commit hooks
  autofix_prs: true
  autoupdate_branch: ''
  autoupdate_commit_msg: '[pre-commit.ci] pre-commit autoupdate'
  autoupdate_schedule: weekly
  skip: [mypy]  # mypy needs full project context

Installation and usage:

# Install pre-commit
pip install pre-commit

# Install git hooks (run once per repo)
pre-commit install
pre-commit install --hook-type commit-msg

# Run all hooks manually
pre-commit run --all-files

# Update hooks to latest versions
pre-commit autoupdate

Key points:

• Hooks run automatically on git commit
• --fix flag enables auto-correction
• Include security scanning (Bandit)
• Use no-commit-to-branch to protect main

---

Example 3: CI/CD Quality Gates

Use case: Enforce quality standards in GitHub Actions.

# .github/workflows/quality.yml

name: Code Quality

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  lint:
    name: Lint and Format
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install ruff

      - name: Run Ruff linter
        run: ruff check --output-format=github .

      - name: Run Ruff formatter
        run: ruff format --check .

  type-check:
    name: Type Check
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install poetry
          poetry install --with dev

      - name: Run mypy
        run: poetry run mypy src/

  test:
    name: Test with Coverage
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install poetry
          poetry install --with dev

      - name: Run tests with coverage
        run: |
          poetry run pytest --cov=src --cov-report=xml --cov-fail-under=80

      - name: Upload coverage reports
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml
          fail_ci_if_error: true

  security:
    name: Security Scan
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install Bandit
        run: pip install bandit[toml]

      - name: Run Bandit
        run: bandit -c pyproject.toml -r src/

Bandit configuration in pyproject.toml:

# pyproject.toml - Security scanning

[tool.bandit]
targets = ["src"]
exclude_dirs = ["tests", ".venv"]
skips = ["B101"]  # Skip assert warnings

Key points:

• Run linting, type checking, and tests in parallel jobs
• Use --output-format=github for inline annotations
• Set --cov-fail-under=80 to enforce coverage threshold
• Include security scanning as a quality gate

---

Example 4: Measuring Code Quality

Use case: Track complexity and maintainability metrics.

# Install quality analysis tools
pip install radon xenon

# Calculate cyclomatic complexity
# Shows functions with complexity >= C (Complex)
radon cc src/ --min C --show-complexity

# Calculate maintainability index
# Score: A (high) to F (low maintainability)
radon mi src/ --show

# Fail build if complexity too high
xenon src/ --max-absolute C --max-modules B --max-average B

Python script for quality metrics:

#!/usr/bin/env python3
"""Quality metrics collection and reporting."""

import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path


@dataclass
class QualityReport:
    """Quality metrics report."""

    complexity_issues: int
    maintainability_avg: float
    test_coverage: float
    lint_errors: int


def run_command(cmd: list[str]) -> tuple[int, str]:
    """Run a command and return exit code and output."""
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.returncode, result.stdout + result.stderr


def check_complexity(src_path: str = "src") -> int:
    """Check cyclomatic complexity, return count of complex functions."""
    code, output = run_command(
        ["radon", "cc", src_path, "--min", "C", "--json"]
    )
    if code != 0 or not output.strip():
        return 0

    import json
    data = json.loads(output)
    return sum(len(funcs) for funcs in data.values())


def check_maintainability(src_path: str = "src") -> float:
    """Get average maintainability index."""
    code, output = run_command(
        ["radon", "mi", src_path, "--json"]
    )
    if code != 0 or not output.strip():
        return 0.0

    import json
    data = json.loads(output)
    scores = [info["mi"] for info in data.values()]
    return sum(scores) / len(scores) if scores else 0.0


def check_lint(src_path: str = ".") -> int:
    """Run Ruff and return error count."""
    code, output = run_command(["ruff", "check", src_path, "--quiet"])
    return output.count("\n") if output else 0


def check_coverage() -> float:
    """Get test coverage percentage."""
    code, output = run_command(
        ["pytest", "--cov=src", "--cov-report=term", "-q"]
    )
    # Parse "TOTAL ... XX%" from output
    for line in output.split("\n"):
        if "TOTAL" in line and "%" in line:
            parts = line.split()
            for part in parts:
                if part.endswith("%"):
                    return float(part[:-1])
    return 0.0


def main() -> int:
    """Generate quality report and check thresholds."""
    print("=== Code Quality Report ===\n")

    report = QualityReport(
        complexity_issues=check_complexity(),
        maintainability_avg=check_maintainability(),
        test_coverage=check_coverage(),
        lint_errors=check_lint(),
    )

    print(f"Complexity Issues (C+): {report.complexity_issues}")
    print(f"Maintainability Index:  {report.maintainability_avg:.1f}")
    print(f"Test Coverage:          {report.test_coverage:.1f}%")
    print(f"Lint Errors:            {report.lint_errors}")

    # Quality gates
    failed = False

    if report.complexity_issues > 5:
        print("\n❌ Too many complex functions")
        failed = True

    if report.maintainability_avg < 65:
        print("\n❌ Maintainability below threshold")
        failed = True

    if report.test_coverage < 80:
        print("\n❌ Test coverage below 80%")
        failed = True

    if report.lint_errors > 0:
        print("\n❌ Lint errors found")
        failed = True

    if not failed:
        print("\n✅ All quality gates passed!")

    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())

Key points:

• Cyclomatic complexity: aim for < 10 per function
• Maintainability index: A (100-20), B (19-10), C (9-0)
• Set quality gates to fail builds on threshold violations

---

Common Patterns

Pattern: Makefile Quality Commands

.PHONY: lint format check

lint:
	ruff check .
	mypy src/

format:
	ruff format .
	ruff check --fix .

check: lint
	pytest --cov=src --cov-fail-under=80

Pattern: VS Code Settings

{
    "[python]": {
        "editor.formatOnSave": true,
        "editor.codeActionsOnSave": {
            "source.fixAll": "explicit",
            "source.organizeImports": "explicit"
        },
        "editor.defaultFormatter": "charliermarsh.ruff"
    },
    "ruff.lint.run": "onSave",
    "mypy-type-checker.args": ["--strict"]
}

Pattern: Editor Integration

# Install VS Code extensions
code --install-extension charliermarsh.ruff
code --install-extension ms-python.mypy-type-checker
code --install-extension ms-python.python

---

Pitfalls to Avoid

Don't do this:

# Suppressing warnings without understanding them
# type: ignore
# noqa
# pylint: disable=all

Do this instead:

# Be specific about what you're ignoring and why
result = some_function()  # type: ignore[return-value]  # TODO: Fix #123
code = "OK"  # noqa: E501 - URL too long to split

---

Don't do this:

# Disabling pre-commit on merge
SKIP=ruff,mypy git commit -m "Quick fix"

Do this instead:

# Fix the issues or use proper escape hatch
git commit -m "WIP: temporary" --no-verify  # Only for true WIP
# Then clean up before merge

---

See Also

• project-structure.md - Project organization
• ci-cd-pipelines.md - Full CI/CD setup
• pytest-essentials.md - Testing integration

Modern Project Structure and Organization

Contents

• Quick Snippets
• Core Concepts
• Production Examples
• Example 1: Standard Project Layout
• Example 2: Configuration with Pydantic Settings
• Example 3: Entry Points and CLI
• Example 4: pyproject.toml Configuration
• Common Patterns
• Pitfalls to Avoid
• See Also

---

Quick Snippets

---

Core Concepts

A well-structured Python project separates concerns into distinct areas:

• src/ layout isolates package code from project files
• tests/ keeps tests separate from implementation
• pyproject.toml centralizes all configuration (PEP 517/518)
• Configuration lives in .env files or environment variables

Benefits of proper structure:

• Reduced onboarding time for new developers
• Improved testability with clear boundaries
• Simplified CI/CD automation
• Future-proofing for tool evolution

---

Production Examples

Example 1: Standard Project Layout

Use case: Organize a Python package for maintainability and distribution.

myproject/
├── src/                      # Source code directory (src layout)
│   └── myproject/            # Main package
│       ├── __init__.py       # Package initialization
│       ├── __main__.py       # Enables: python -m myproject
│       ├── cli.py            # Command-line interface
│       ├── config.py         # Configuration management
│       ├── core/             # Core business logic
│       │   ├── __init__.py
│       │   ├── models.py
│       │   └── services.py
│       └── utils/            # Utility functions
│           ├── __init__.py
│           └── helpers.py
├── tests/                    # Test directory (mirrors src structure)
│   ├── __init__.py
│   ├── conftest.py           # Shared pytest fixtures
│   ├── test_cli.py
│   └── core/
│       ├── __init__.py
│       └── test_models.py
├── docs/                     # Documentation
│   └── README.md
├── scripts/                  # Development/deployment scripts
│   └── setup_dev.sh
├── .env.example              # Example environment variables
├── .gitignore
├── LICENSE
├── README.md
├── pyproject.toml            # Project configuration (PEP 517/518)
└── Makefile                  # Common development tasks

Package `__init__.py`:

#!/usr/bin/env python3
"""MyProject - A well-structured Python package.

This module exposes the public API of the package.
"""

from myproject.core.models import User, Order
from myproject.core.services import process_order

__version__ = "0.1.0"
__all__ = ["User", "Order", "process_order", "__version__"]

Package `__main__.py`:

#!/usr/bin/env python3
"""Enable running the package with: python -m myproject"""

from myproject.cli import main

if __name__ == "__main__":
    main()

Key points:

• The src/ layout prevents accidental imports from project root
• __all__ explicitly defines the public API
• __main__.py enables python -m myproject execution

---

Example 2: Configuration with Pydantic Settings

Use case: Type-safe configuration from environment variables and .env files.

#!/usr/bin/env python3
"""Configuration management with Pydantic Settings v2.

Loads settings from environment variables with validation,
type conversion, and sensible defaults.
"""

from functools import lru_cache
from typing import Literal

from pydantic import Field, SecretStr, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


class DatabaseSettings(BaseSettings):
    """Database connection settings."""

    model_config = SettingsConfigDict(
        env_prefix="DB_",
        env_file=".env",
        env_file_encoding="utf-8",
    )

    url: str = "sqlite:///./app.db"
    pool_size: int = Field(default=5, ge=1, le=100)
    pool_timeout: float = Field(default=30.0, gt=0)
    echo: bool = False  # Log SQL queries


class AppSettings(BaseSettings):
    """Application-wide settings."""

    model_config = SettingsConfigDict(
        env_prefix="APP_",
        env_file=".env",
        env_file_encoding="utf-8",
    )

    # Application info
    name: str = "MyProject"
    version: str = "0.1.0"
    debug: bool = False

    # Environment
    environment: Literal["development", "staging", "production"] = "development"

    # Security
    secret_key: SecretStr
    api_key: SecretStr | None = None

    # Logging
    log_level: str = Field(default="INFO")

    @field_validator("log_level")
    @classmethod
    def validate_log_level(cls, v: str) -> str:
        """Ensure log level is valid."""
        valid_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
        upper_v = v.upper()
        if upper_v not in valid_levels:
            raise ValueError(f"log_level must be one of {valid_levels}")
        return upper_v


class Settings(BaseSettings):
    """Root settings combining all configuration sections."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",  # Ignore extra env vars
    )

    app: AppSettings = Field(default_factory=AppSettings)
    database: DatabaseSettings = Field(default_factory=DatabaseSettings)


@lru_cache
def get_settings() -> Settings:
    """Get cached settings instance.

    Using lru_cache ensures settings are only loaded once,
    improving performance and consistency.
    """
    return Settings()


# Usage example
if __name__ == "__main__":
    settings = get_settings()

    print(f"App: {settings.app.name} v{settings.app.version}")
    print(f"Environment: {settings.app.environment}")
    print(f"Debug: {settings.app.debug}")
    print(f"Log Level: {settings.app.log_level}")
    print(f"Database URL: {settings.database.url}")
    print(f"Pool Size: {settings.database.pool_size}")

    # SecretStr hides value in output
    print(f"Secret Key: {settings.app.secret_key}")
    # To get the actual value:
    # settings.app.secret_key.get_secret_value()

Example `.env` file:

# .env - Environment-specific settings (DO NOT COMMIT)
APP_ENVIRONMENT=development
APP_DEBUG=true
APP_SECRET_KEY=your-secret-key-here
APP_LOG_LEVEL=DEBUG

DB_URL=postgresql://user:pass@localhost:5432/mydb
DB_POOL_SIZE=10
DB_ECHO=true

Example `.env.example` file:

# .env.example - Template for required environment variables
# Copy to .env and fill in values

APP_ENVIRONMENT=development
APP_DEBUG=false
APP_SECRET_KEY=generate-a-secure-key
APP_LOG_LEVEL=INFO

DB_URL=sqlite:///./app.db
DB_POOL_SIZE=5

Key points:

• Use @lru_cache to avoid reloading settings
• SecretStr hides sensitive values from logs
• Prefix env vars by section (APP_, DB_) for organization
• Provide .env.example as a template (commit this, not .env)

---

Example 3: Entry Points and CLI

Use case: Create installable command-line tools with Click.

#!/usr/bin/env python3
"""Command-line interface using Click.

Provides a clean CLI with subcommands, options, and help text.
"""

import sys
from pathlib import Path

import click

from myproject import __version__
from myproject.config import get_settings
from myproject.core.services import process_data


@click.group()
@click.version_option(version=__version__, prog_name="myproject")
@click.option("--verbose", "-v", is_flag=True, help="Enable verbose output")
@click.pass_context
def cli(ctx: click.Context, verbose: bool) -> None:
    """MyProject - A tool for processing data.

    Run 'myproject COMMAND --help' for command-specific help.
    """
    ctx.ensure_object(dict)
    ctx.obj["verbose"] = verbose
    ctx.obj["settings"] = get_settings()


@cli.command()
@click.argument("input_file", type=click.Path(exists=True, path_type=Path))
@click.option(
    "--output", "-o",
    type=click.Path(path_type=Path),
    default=None,
    help="Output file path (default: stdout)"
)
@click.option(
    "--format", "-f",
    type=click.Choice(["json", "csv", "yaml"]),
    default="json",
    help="Output format"
)
@click.pass_context
def process(
    ctx: click.Context,
    input_file: Path,
    output: Path | None,
    format: str
) -> None:
    """Process an input file and generate output.

    INPUT_FILE: Path to the file to process
    """
    verbose = ctx.obj["verbose"]

    if verbose:
        click.echo(f"Processing: {input_file}")
        click.echo(f"Format: {format}")

    try:
        result = process_data(input_file, output_format=format)

        if output:
            output.write_text(result)
            click.echo(f"Output written to: {output}")
        else:
            click.echo(result)

    except Exception as e:
        click.echo(f"Error: {e}", err=True)
        sys.exit(1)


@cli.command()
@click.pass_context
def config(ctx: click.Context) -> None:
    """Display current configuration."""
    settings = ctx.obj["settings"]

    click.echo("Current Configuration:")
    click.echo(f"  Environment: {settings.app.environment}")
    click.echo(f"  Debug: {settings.app.debug}")
    click.echo(f"  Log Level: {settings.app.log_level}")
    click.echo(f"  Database: {settings.database.url}")


@cli.command()
@click.option("--check", is_flag=True, help="Check health without fixing")
@click.pass_context
def health(ctx: click.Context, check: bool) -> None:
    """Check application health status."""
    verbose = ctx.obj["verbose"]

    checks = [
        ("Configuration", True),
        ("Database", True),
        ("External Services", True),
    ]

    all_ok = True
    for name, status in checks:
        icon = "✓" if status else "✗"
        color = "green" if status else "red"
        click.echo(click.style(f"  {icon} {name}", fg=color))
        if not status:
            all_ok = False

    if all_ok:
        click.echo(click.style("\nAll checks passed!", fg="green"))
    else:
        click.echo(click.style("\nSome checks failed!", fg="red"))
        sys.exit(1)


def main() -> None:
    """Entry point for the CLI."""
    cli()


if __name__ == "__main__":
    main()

Key points:

• Use @click.group() for subcommands
• Pass context with @click.pass_context for shared state
• Use click.Path(path_type=Path) for typed path arguments
• Exit with non-zero status on errors

---

Example 4: pyproject.toml Configuration

Use case: Centralize all project configuration in a single file.

# pyproject.toml - Project configuration (PEP 517/518)

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

[tool.poetry]
name = "myproject"
version = "0.1.0"
description = "A well-structured Python project"
authors = ["Your Name <you@example.com>"]
license = "MIT"
readme = "README.md"
homepage = "https://github.com/username/myproject"
repository = "https://github.com/username/myproject"
documentation = "https://myproject.readthedocs.io"
keywords = ["python", "example", "project"]
classifiers = [
    "Development Status :: 3 - Alpha",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: MIT License",
    "Programming Language :: Python :: 3.12",
]
packages = [{ include = "myproject", from = "src" }]

[tool.poetry.dependencies]
python = "^3.12"
click = "^8.1.0"
pydantic = "^2.5.0"
pydantic-settings = "^2.1.0"

[tool.poetry.group.dev.dependencies]
pytest = "^8.0.0"
pytest-cov = "^4.1.0"
ruff = "^0.1.0"
mypy = "^1.7.0"
pre-commit = "^3.6.0"

[tool.poetry.scripts]
myproject = "myproject.cli:main"

# Ruff configuration
[tool.ruff]
line-length = 88
target-version = "py312"
src = ["src", "tests"]

[tool.ruff.lint]
select = [
    "E",     # pycodestyle errors
    "F",     # pyflakes
    "I",     # isort
    "B",     # flake8-bugbear
    "C4",    # flake8-comprehensions
    "UP",    # pyupgrade
    "ARG",   # flake8-unused-arguments
    "SIM",   # flake8-simplify
]
ignore = ["E501"]  # Line too long (handled by formatter)

[tool.ruff.lint.isort]
known-first-party = ["myproject"]

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

# Mypy configuration
[tool.mypy]
python_version = "3.12"
strict = true
warn_return_any = true
warn_unused_configs = true
plugins = ["pydantic.mypy"]

[[tool.mypy.overrides]]
module = "tests.*"
disallow_untyped_defs = false

# Pytest configuration
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_functions = ["test_*"]
addopts = [
    "--strict-markers",
    "-ra",
    "-q",
]
markers = [
    "slow: marks tests as slow",
    "integration: marks integration tests",
]

# Coverage configuration
[tool.coverage.run]
source = ["src"]
branch = true
omit = ["*/tests/*"]

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

Key points:

• packages = [{ include = "myproject", from = "src" }] for src layout
• [tool.poetry.scripts] defines CLI entry points
• Configure all tools (ruff, mypy, pytest, coverage) in one file
• Use dependency groups for dev/test/docs separation

---

Common Patterns

Pattern: Makefile for Common Tasks

.PHONY: install test lint format clean

install:
	poetry install

test:
	poetry run pytest

lint:
	poetry run ruff check .
	poetry run mypy src/

format:
	poetry run ruff format .
	poetry run ruff check --fix .

clean:
	rm -rf .pytest_cache .mypy_cache .ruff_cache
	find . -type d -name __pycache__ -exec rm -rf {} +

Pattern: conftest.py for Shared Fixtures

# tests/conftest.py
import pytest
from myproject.config import Settings

@pytest.fixture
def settings():
    """Provide test settings."""
    return Settings(
        app={"environment": "testing", "debug": True},
        database={"url": "sqlite:///:memory:"},
    )

@pytest.fixture
def client(settings):
    """Provide test client."""
    from myproject.app import create_app
    app = create_app(settings)
    return app.test_client()

---

Pitfalls to Avoid

Don't do this:

# Hardcoded configuration in code
DATABASE_URL = "postgresql://user:password@localhost/db"
API_KEY = "sk_live_abc123"

Do this instead:

# Load from environment with validation
from myproject.config import get_settings

settings = get_settings()
database_url = settings.database.url
api_key = settings.app.api_key.get_secret_value()

---

Don't do this:

# Flat project structure
myproject/
├── main.py
├── utils.py
├── models.py
├── test_main.py      # Tests mixed with code!
└── requirements.txt

Do this instead:

# Proper src layout
myproject/
├── src/myproject/    # Package code
├── tests/            # Separate test directory
└── pyproject.toml    # Modern configuration

---

See Also

• code-quality.md - Linting and formatting setup
• poetry-workflow.md - Dependency management
• pyproject-config.md - Deep dive into pyproject.toml

Python Syntax Essentials

Contents

• Quick Snippets
• Core Concepts
• Production Examples
• Example 1: Modern String Formatting
• Example 2: Comprehensions and Generators
• Example 3: Structural Pattern Matching
• Example 4: Walrus Operator and Modern Syntax
• Common Patterns
• Pitfalls to Avoid
• See Also

---

Quick Snippets

---

Core Concepts

Modern Python (3.10+) provides expressive syntax for common operations:

• f-strings for readable string interpolation with expressions
• Comprehensions for creating collections declaratively
• Pattern matching (3.10+) for structural decomposition
• Walrus operator (3.8+) for assignment expressions

Python emphasizes readability and explicit over implicit behavior. The Zen of Python (import this) guides idiomatic code: prefer flat over nested, simple over complex.

---

Production Examples

Example 1: Modern String Formatting

Use case: Format strings clearly and safely in production code.

#!/usr/bin/env python3
"""Modern string formatting with f-strings."""

from datetime import datetime
from decimal import Decimal


def format_examples() -> None:
    """Demonstrate f-string capabilities."""
    name = "Alice"
    age = 30
    balance = Decimal("1234.56")
    items = ["apple", "banana", "cherry"]

    # Basic interpolation
    print(f"Name: {name}, Age: {age}")

    # Expression evaluation
    print(f"Age in months: {age * 12}")

    # Method calls
    print(f"Uppercase: {name.upper()}")

    # Debug mode (Python 3.8+) - prints variable name and value
    print(f"{name=}, {age=}")
    # Output: name='Alice', age=30

    # Formatting specifications
    print(f"Balance: ${balance:,.2f}")  # $1,234.56
    print(f"Percentage: {0.856:.1%}")   # 85.6%
    print(f"Padded: {age:05d}")         # 00030
    print(f"Left align: {name:<10}|")   # Alice     |
    print(f"Center: {name:^10}|")       # Alice  |

    # Date formatting
    now = datetime.now()
    print(f"Date: {now:%Y-%m-%d %H:%M}")

    # Conditional expressions
    status = "adult" if age >= 18 else "minor"
    print(f"Status: {status}")

    # Multi-line f-strings
    report = f"""
    User Report
    -----------
    Name:    {name}
    Age:     {age}
    Balance: ${balance:,.2f}
    Items:   {len(items)}
    """
    print(report)


def format_numbers() -> None:
    """Number formatting patterns."""
    value = 1234567.89

    # Thousands separator
    print(f"Comma: {value:,.2f}")       # 1,234,567.89
    print(f"Underscore: {value:_.2f}")  # 1_234_567.89

    # Scientific notation
    print(f"Scientific: {value:.2e}")   # 1.23e+06

    # Binary, octal, hex
    n = 255
    print(f"Binary: {n:b}")   # 11111111
    print(f"Octal: {n:o}")    # 377
    print(f"Hex: {n:x}")      # ff
    print(f"Hex upper: {n:X}")  # FF


def safe_string_formatting(user_input: str) -> str:
    """Safely format strings with user input.

    Never use .format() or % with untrusted input!
    f-strings are safer because expressions are evaluated
    at definition time, not runtime.
    """
    # Safe: f-string with escaped output
    return f"User said: {user_input!r}"


if __name__ == "__main__":
    format_examples()
    format_numbers()
    print(safe_string_formatting("Hello\nWorld"))

Key points:

• Use {var=} for debug output showing name and value
• Use format specs like :,.2f for numbers
• f-strings are evaluated at definition, safer than .format()

---

Example 2: Comprehensions and Generators

Use case: Create collections and iterators declaratively.

#!/usr/bin/env python3
"""Comprehensions and generators for declarative data transformation."""

from collections.abc import Generator, Iterator
from typing import Any


def list_comprehensions() -> None:
    """List comprehension patterns."""
    numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

    # Basic transformation
    doubled = [x * 2 for x in numbers]
    print(f"Doubled: {doubled}")

    # With filter
    evens = [x for x in numbers if x % 2 == 0]
    print(f"Evens: {evens}")

    # Combined transformation and filter
    doubled_evens = [x * 2 for x in numbers if x % 2 == 0]
    print(f"Doubled evens: {doubled_evens}")

    # Nested loops (flatten)
    matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
    flat = [x for row in matrix for x in row]
    print(f"Flattened: {flat}")

    # Conditional expression in output
    labels = ["even" if x % 2 == 0 else "odd" for x in numbers[:5]]
    print(f"Labels: {labels}")


def dict_comprehensions() -> None:
    """Dictionary comprehension patterns."""
    words = ["apple", "banana", "cherry"]

    # Create dict from list
    word_lengths = {word: len(word) for word in words}
    print(f"Lengths: {word_lengths}")

    # Filter and transform
    items = {"a": 1, "b": 2, "c": 3, "d": 4}
    filtered = {k: v * 10 for k, v in items.items() if v > 1}
    print(f"Filtered: {filtered}")

    # Swap keys and values
    swapped = {v: k for k, v in items.items()}
    print(f"Swapped: {swapped}")

    # From two lists with zip
    keys = ["name", "age", "city"]
    values = ["Alice", 30, "NYC"]
    combined = dict(zip(keys, values))
    print(f"Combined: {combined}")


def set_comprehensions() -> None:
    """Set comprehension patterns."""
    words = ["hello", "world", "hello", "python"]

    # Unique first letters
    first_letters = {word[0] for word in words}
    print(f"First letters: {first_letters}")

    # Unique lengths
    lengths = {len(word) for word in words}
    print(f"Unique lengths: {lengths}")


def generator_expressions() -> None:
    """Generator expressions for memory efficiency."""
    # Generator expression (lazy evaluation)
    squares_gen = (x**2 for x in range(1000000))
    print(f"Generator: {squares_gen}")
    print(f"First 5: {[next(squares_gen) for _ in range(5)]}")

    # Sum without creating intermediate list
    total = sum(x**2 for x in range(1000))
    print(f"Sum of squares: {total}")

    # any() and all() with generators
    numbers = [2, 4, 6, 8, 10]
    all_even = all(x % 2 == 0 for x in numbers)
    any_greater = any(x > 5 for x in numbers)
    print(f"All even: {all_even}, Any > 5: {any_greater}")


def generator_function() -> Generator[int, None, None]:
    """Generator function with yield.

    Yields:
        Fibonacci numbers indefinitely
    """
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b


def take(n: int, iterable: Iterator[Any]) -> list[Any]:
    """Take first n items from an iterator."""
    return [next(iterable) for _ in range(n)]


def pipeline_example() -> None:
    """Demonstrate generator pipelines."""

    def numbers() -> Generator[int, None, None]:
        """Generate numbers 1 to infinity."""
        n = 1
        while True:
            yield n
            n += 1

    def square(nums: Iterator[int]) -> Generator[int, None, None]:
        """Square each number."""
        for n in nums:
            yield n ** 2

    def filter_even(nums: Iterator[int]) -> Generator[int, None, None]:
        """Keep only even numbers."""
        for n in nums:
            if n % 2 == 0:
                yield n

    # Compose pipeline (lazy - nothing computed yet)
    pipeline = filter_even(square(numbers()))

    # Only computes what's needed
    result = take(5, pipeline)
    print(f"Pipeline result: {result}")  # [4, 16, 36, 64, 100]


if __name__ == "__main__":
    list_comprehensions()
    print()
    dict_comprehensions()
    print()
    set_comprehensions()
    print()
    generator_expressions()
    print()
    fib = generator_function()
    print(f"First 10 Fibonacci: {take(10, fib)}")
    print()
    pipeline_example()

Key points:

• Use list comprehensions for small-to-medium transformations
• Use generators for large data or infinite sequences
• Generators enable memory-efficient pipelines
• Prefer sum(x for x in ...) over sum([x for x in ...])

---

Example 3: Structural Pattern Matching

Use case: Match and destructure complex data structures (Python 3.10+).

#!/usr/bin/env python3
"""Structural pattern matching (Python 3.10+)."""

from dataclasses import dataclass
from typing import Any


@dataclass
class Point:
    """A 2D point."""
    x: float
    y: float


@dataclass
class Circle:
    """A circle shape."""
    center: Point
    radius: float


@dataclass
class Rectangle:
    """A rectangle shape."""
    top_left: Point
    width: float
    height: float


def describe_point(point: Point) -> str:
    """Describe a point's location using pattern matching."""
    match point:
        case Point(x=0, y=0):
            return "Origin"
        case Point(x=0, y=y):
            return f"On Y-axis at y={y}"
        case Point(x=x, y=0):
            return f"On X-axis at x={x}"
        case Point(x=x, y=y) if x == y:
            return f"On diagonal at ({x}, {y})"
        case Point(x=x, y=y):
            return f"Point at ({x}, {y})"


def calculate_area(shape: Circle | Rectangle) -> float:
    """Calculate area using pattern matching on types."""
    import math

    match shape:
        case Circle(radius=r):
            return math.pi * r ** 2
        case Rectangle(width=w, height=h):
            return w * h
        case _:
            raise ValueError(f"Unknown shape: {shape}")


def process_command(command: dict[str, Any]) -> str:
    """Process commands using pattern matching on dict structure."""
    match command:
        case {"action": "quit"}:
            return "Goodbye!"
        case {"action": "greet", "name": name}:
            return f"Hello, {name}!"
        case {"action": "add", "x": x, "y": y}:
            return f"Result: {x + y}"
        case {"action": "list", "items": [first, *rest]}:
            return f"First: {first}, remaining: {len(rest)}"
        case {"action": action}:
            return f"Unknown action: {action}"
        case _:
            return "Invalid command"


def process_http_response(response: tuple[int, str]) -> str:
    """Match HTTP response codes."""
    match response:
        case (200, body):
            return f"Success: {body[:50]}..."
        case (201, _):
            return "Created"
        case (204, _):
            return "No content"
        case (301 | 302, _):
            return "Redirect"
        case (400, _):
            return "Bad request"
        case (401 | 403, _):
            return "Auth error"
        case (404, _):
            return "Not found"
        case (status, _) if 500 <= status < 600:
            return f"Server error: {status}"
        case (status, _):
            return f"Unknown status: {status}"


def parse_json_event(event: dict[str, Any]) -> str:
    """Parse JSON events with nested patterns."""
    match event:
        case {
            "type": "user",
            "action": "login",
            "data": {"user_id": uid, "timestamp": ts}
        }:
            return f"User {uid} logged in at {ts}"

        case {
            "type": "order",
            "action": "created",
            "data": {"order_id": oid, "items": [_, *_] as items}
        }:
            return f"Order {oid} created with {len(items)} items"

        case {"type": t, "action": a}:
            return f"Event: {t}/{a}"

        case _:
            return "Unknown event format"


if __name__ == "__main__":
    # Point matching
    points = [
        Point(0, 0),
        Point(0, 5),
        Point(3, 0),
        Point(4, 4),
        Point(3, 7),
    ]
    for p in points:
        print(f"{p} -> {describe_point(p)}")

    print()

    # Shape matching
    shapes = [
        Circle(Point(0, 0), 5),
        Rectangle(Point(0, 0), 4, 3),
    ]
    for shape in shapes:
        print(f"{shape} -> area = {calculate_area(shape):.2f}")

    print()

    # Command matching
    commands = [
        {"action": "greet", "name": "Alice"},
        {"action": "add", "x": 5, "y": 3},
        {"action": "list", "items": [1, 2, 3, 4]},
        {"action": "unknown"},
    ]
    for cmd in commands:
        print(f"{cmd} -> {process_command(cmd)}")

    print()

    # HTTP response matching
    responses = [
        (200, "OK data"),
        (404, "Not found"),
        (500, "Error"),
    ]
    for resp in responses:
        print(f"{resp[0]} -> {process_http_response(resp)}")

Key points:

• Pattern matching replaces complex if/elif chains
• Use | for OR patterns, guards for conditions
• Capture values with variable names in patterns
• _ matches anything without binding

---

Example 4: Walrus Operator and Modern Syntax

Use case: Write concise code with assignment expressions and modern features.

#!/usr/bin/env python3
"""Modern Python syntax features (3.8+)."""

import re
from pathlib import Path


def walrus_examples() -> None:
    """Walrus operator (:=) for assignment expressions."""

    # Read and check in one expression
    data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

    # Without walrus: need separate lines
    # n = len(data)
    # if n > 5:
    #     print(f"Long list: {n} items")

    # With walrus: combine assignment and condition
    if (n := len(data)) > 5:
        print(f"Long list: {n} items")

    # In list comprehensions
    results = [y for x in data if (y := x ** 2) > 10]
    print(f"Squares > 10: {results}")

    # In while loops
    lines = ["line 1", "line 2", "line 3", ""]
    index = 0
    while (line := lines[index] if index < len(lines) else ""):
        print(f"Processing: {line}")
        index += 1

    # With regex
    text = "Contact: alice@example.com or bob@test.org"
    pattern = r"[\w\.-]+@[\w\.-]+\.\w+"
    if match := re.search(pattern, text):
        print(f"Found email: {match.group()}")

    # Find all with any()
    numbers = [1, 3, 5, 8, 9]
    if any((even := n) % 2 == 0 for n in numbers):
        print(f"First even: {even}")


def unpacking_examples() -> None:
    """Extended unpacking with * operator."""

    # Basic unpacking
    first, second, third = [1, 2, 3]

    # Capture rest with *
    first, *middle, last = [1, 2, 3, 4, 5]
    print(f"First: {first}, Middle: {middle}, Last: {last}")
    # First: 1, Middle: [2, 3, 4], Last: 5

    # Head and tail
    head, *tail = [1, 2, 3, 4]
    print(f"Head: {head}, Tail: {tail}")

    # Ignore values with _
    name, _, age = ("Alice", "ignored", 30)

    # Nested unpacking
    data = [("Alice", 30), ("Bob", 25)]
    for name, age in data:
        print(f"{name} is {age}")

    # Dict merging (3.9+)
    defaults = {"color": "red", "size": "medium"}
    overrides = {"size": "large", "quantity": 5}
    merged = {**defaults, **overrides}
    print(f"Merged: {merged}")

    # Dict union operator (3.9+)
    merged2 = defaults | overrides
    print(f"Union: {merged2}")


def positional_only_params(x: int, y: int, /, *, keyword: str) -> str:
    """Function with positional-only and keyword-only params.

    Args:
        x: First positional-only parameter
        y: Second positional-only parameter
        keyword: Keyword-only parameter

    The / marks end of positional-only params.
    The * marks start of keyword-only params.
    """
    return f"x={x}, y={y}, keyword={keyword}"


def modern_file_handling() -> None:
    """Modern file handling with pathlib and walrus."""
    from tempfile import NamedTemporaryFile

    # Create temp file for demo
    with NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
        f.write("Line 1\nLine 2\nLine 3\n")
        temp_path = Path(f.name)

    # Read with walrus for EOF check
    with open(temp_path) as f:
        while line := f.readline():
            print(f"Read: {line.strip()}")

    # Pathlib operations
    if temp_path.exists():
        print(f"File size: {temp_path.stat().st_size} bytes")
        print(f"Suffix: {temp_path.suffix}")
        temp_path.unlink()  # Delete


def exception_groups() -> None:
    """Exception groups (Python 3.11+)."""

    def process_items(items: list[str]) -> list[str]:
        """Process items, collecting all errors."""
        results = []
        errors = []

        for item in items:
            try:
                if item.startswith("bad"):
                    raise ValueError(f"Invalid item: {item}")
                results.append(item.upper())
            except ValueError as e:
                errors.append(e)

        if errors:
            raise ExceptionGroup("Processing errors", errors)

        return results

    try:
        process_items(["good", "bad1", "ok", "bad2"])
    except* ValueError as eg:
        print(f"Caught {len(eg.exceptions)} ValueError(s)")
        for e in eg.exceptions:
            print(f"  - {e}")


if __name__ == "__main__":
    print("=== Walrus Operator ===")
    walrus_examples()

    print("\n=== Unpacking ===")
    unpacking_examples()

    print("\n=== Positional-Only Params ===")
    result = positional_only_params(1, 2, keyword="test")
    print(result)

    print("\n=== File Handling ===")
    modern_file_handling()

    print("\n=== Exception Groups ===")
    exception_groups()

Key points:

• Walrus operator (:=) enables inline assignment
• Use / for positional-only, * for keyword-only params
• Extended unpacking with *rest captures remaining items
• Dict union | merges dicts (3.9+)

---

Common Patterns

Pattern: Guard Clauses

def process(data: list[int] | None) -> int:
    # Early return for edge cases
    if data is None:
        return 0
    if not data:
        return 0
    if len(data) == 1:
        return data[0]

    # Main logic after guards
    return sum(data) // len(data)

Pattern: EAFP (Easier to Ask Forgiveness)

# Pythonic: try/except (EAFP)
try:
    value = my_dict["key"]
except KeyError:
    value = "default"

# Also good: .get() for dicts
value = my_dict.get("key", "default")

Pattern: Context Variables

from contextvars import ContextVar

request_id: ContextVar[str] = ContextVar("request_id", default="unknown")

def log(message: str) -> None:
    print(f"[{request_id.get()}] {message}")

---

Pitfalls to Avoid

Don't do this:

# Mutable default argument (shared between calls!)
def append_to(item, target=[]):
    target.append(item)
    return target

Do this instead:

def append_to(item, target=None):
    if target is None:
        target = []
    target.append(item)
    return target

---

Don't do this:

# Using type() for type checking
if type(x) == list:
    ...

Do this instead:

# Use isinstance for proper type checking
if isinstance(x, list):
    ...
# Or for ABCs
from collections.abc import Sequence
if isinstance(x, Sequence):
    ...

---

See Also

• type-systems.md - Type hints and annotations
• generators.md - Generator patterns in depth
• error-handling.md - Exception handling

Type Hints and Static Analysis

Contents

• Quick Snippets
• Core Concepts
• Production Examples
• Example 1: Basic Type Annotations
• Example 2: Generic Types with Python 3.12+
• Example 3: Protocols for Structural Typing
• Example 4: Pydantic Runtime Validation
• Common Patterns
• Pitfalls to Avoid
• See Also

---

Quick Snippets

---

Core Concepts

Type hints provide optional static typing while maintaining Python's dynamic nature. They enable:

• Early error detection via tools like mypy and pyright
• Better IDE support with intelligent autocompletion
• Self-documenting code that clarifies function contracts

Python 3.9+ allows built-in types as generics (list[str] instead of List[str]). Python 3.10+ introduced the | operator for unions (int | None instead of Optional[int]). Python 3.12+ adds type parameter syntax for generics and the type statement for aliases.

---

Production Examples

Example 1: Basic Type Annotations

Use case: Annotate variables, functions, and classes for clarity and validation.

#!/usr/bin/env python3
"""Basic type annotation patterns for Python 3.12+."""

from typing import Optional

# Variable annotations
count: int = 0
name: str = "Alice"
is_active: bool = True
temperature: float = 98.6

# Function with full type annotations
def calculate_average(numbers: list[float]) -> float:
    """Calculate the average of a list of numbers.

    Args:
        numbers: List of numeric values

    Returns:
        The arithmetic mean of the input values
    """
    if not numbers:
        return 0.0
    return sum(numbers) / len(numbers)


# Function with optional parameter
def greet(name: str, title: str | None = None) -> str:
    """Generate a greeting message.

    Args:
        name: The person's name
        title: Optional title (Mr., Ms., Dr., etc.)

    Returns:
        Formatted greeting string
    """
    if title is not None:
        return f"Hello, {title} {name}!"
    return f"Hello, {name}!"


# Class with type annotations
class User:
    """Represents a system user with typed attributes."""

    default_role: str = "guest"  # Class variable

    def __init__(self, name: str, age: int) -> None:
        self.name: str = name
        self.age: int = age
        self.active: bool = True

    def deactivate(self) -> None:
        """Mark the user as inactive."""
        self.active = False


if __name__ == "__main__":
    # Usage examples
    avg = calculate_average([1.0, 2.0, 3.0, 4.0, 5.0])
    print(f"Average: {avg}")

    print(greet("Alice"))
    print(greet("Smith", "Dr."))

    user = User("Bob", 30)
    print(f"User: {user.name}, Active: {user.active}")

Key points:

• Use lowercase built-in types: list, dict, tuple, set
• Use | for union types instead of Union[]
• Return None explicitly for functions that return nothing

---

Example 2: Generic Types with Python 3.12+

Use case: Create reusable, type-safe containers and functions.

#!/usr/bin/env python3
"""Generic types with Python 3.12+ type parameter syntax."""

from typing import TypeAlias

# Python 3.12+ type alias declaration
type Point = tuple[float, float]
type UserData = dict[str, str | int | bool]
type Matrix = list[list[float]]


# Python 3.12+ generic class syntax
class Box[T]:
    """A generic container that holds a single value.

    Type Parameters:
        T: The type of value this box contains
    """

    def __init__(self, content: T) -> None:
        self._content = content

    def get(self) -> T:
        """Retrieve the contained value."""
        return self._content

    def set(self, value: T) -> None:
        """Replace the contained value."""
        self._content = value


# Generic function with type parameter
def first[T](items: list[T]) -> T | None:
    """Return the first item from a list, or None if empty.

    Type Parameters:
        T: The type of items in the list

    Args:
        items: A list of items

    Returns:
        The first item, or None if the list is empty
    """
    return items[0] if items else None


# Multiple type parameters
class Pair[K, V]:
    """A generic key-value pair container.

    Type Parameters:
        K: The type of the key
        V: The type of the value
    """

    def __init__(self, key: K, value: V) -> None:
        self.key = key
        self.value = value

    def swap(self) -> "Pair[V, K]":
        """Create a new Pair with key and value swapped."""
        return Pair(self.value, self.key)


if __name__ == "__main__":
    # Using Box with different types
    int_box: Box[int] = Box(42)
    str_box: Box[str] = Box("hello")

    print(f"Int box: {int_box.get()}")
    print(f"Str box: {str_box.get()}")

    # Using generic function
    numbers = [1, 2, 3]
    names = ["Alice", "Bob"]

    print(f"First number: {first(numbers)}")
    print(f"First name: {first(names)}")

    # Using Pair
    pair: Pair[str, int] = Pair("age", 30)
    swapped = pair.swap()
    print(f"Original: {pair.key}={pair.value}")
    print(f"Swapped: {swapped.key}={swapped.value}")

Key points:

• Python 3.12 class Box[T]: syntax replaces class Box(Generic[T]):
• type statement creates explicit type aliases
• Generic functions use def func[T](...) syntax

---

Example 3: Protocols for Structural Typing

Use case: Define interfaces based on behavior (duck typing) for flexible design.

#!/usr/bin/env python3
"""Protocols enable structural typing - interfaces based on behavior."""

from typing import Protocol, runtime_checkable


class Greeter(Protocol):
    """Protocol defining objects that can greet.

    Any class implementing a `greet(name: str) -> str` method
    satisfies this protocol without explicit inheritance.
    """

    def greet(self, name: str) -> str:
        """Generate a greeting for the given name."""
        ...


class Closeable(Protocol):
    """Protocol for resources that can be closed."""

    def close(self) -> None:
        """Release resources."""
        ...


@runtime_checkable
class Sized(Protocol):
    """Protocol for objects with a length.

    The @runtime_checkable decorator enables isinstance() checks.
    """

    def __len__(self) -> int:
        ...


# Implementations don't need to inherit from protocols
class FriendlyGreeter:
    """A friendly greeter implementation."""

    def greet(self, name: str) -> str:
        return f"Hello, {name}! Nice to meet you!"


class FormalGreeter:
    """A formal greeter implementation."""

    def __init__(self, title: str = "Dear") -> None:
        self.title = title

    def greet(self, name: str) -> str:
        return f"{self.title} {name}, greetings."


# Function accepting any Greeter
def welcome(greeter: Greeter, name: str) -> str:
    """Welcome someone using any greeter implementation.

    Args:
        greeter: Any object with a greet() method
        name: The person to welcome

    Returns:
        The greeting message
    """
    return greeter.greet(name)


# Database-like example with Closeable
class DatabaseConnection:
    """Simulated database connection."""

    def __init__(self, url: str) -> None:
        self.url = url
        self._open = True
        print(f"Connected to {url}")

    def close(self) -> None:
        self._open = False
        print(f"Disconnected from {self.url}")


def use_resource(resource: Closeable) -> None:
    """Use and close any closeable resource."""
    try:
        print("Using resource...")
    finally:
        resource.close()


if __name__ == "__main__":
    # Different greeters work with the same function
    friendly = FriendlyGreeter()
    formal = FormalGreeter("Esteemed")

    print(welcome(friendly, "Alice"))
    print(welcome(formal, "Bob"))

    # Closeable protocol
    db = DatabaseConnection("postgres://localhost/mydb")
    use_resource(db)

    # Runtime checking with @runtime_checkable
    print(f"\nIs [1,2,3] Sized? {isinstance([1, 2, 3], Sized)}")
    print(f"Is 42 Sized? {isinstance(42, Sized)}")

Key points:

• Protocols define structural interfaces without inheritance
• Classes satisfy protocols implicitly by implementing required methods
• @runtime_checkable enables isinstance() checks (use sparingly)

---

Example 4: Pydantic Runtime Validation

Use case: Validate data at runtime using type hints with Pydantic.

#!/usr/bin/env python3
"""Pydantic v2 for runtime type validation and settings management."""

from datetime import date
from decimal import Decimal
from typing import Annotated

from pydantic import (
    BaseModel,
    Field,
    SecretStr,
    field_validator,
    model_validator,
)
from pydantic_settings import BaseSettings


class User(BaseModel):
    """User model with automatic validation.

    Pydantic validates types at runtime and converts compatible values.
    """

    id: int
    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$")
    age: int = Field(..., ge=0, le=150)
    balance: Decimal = Field(default=Decimal("0.00"))

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        """Ensure name is not just whitespace."""
        if not v.strip():
            raise ValueError("name cannot be empty or whitespace")
        return v.strip()


class Order(BaseModel):
    """Order with nested validation and custom validators."""

    order_id: str
    user_id: int
    items: list[str] = Field(..., min_length=1)
    total: Decimal = Field(..., gt=0)
    created_at: date = Field(default_factory=date.today)

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "order_id": "ORD-001",
                    "user_id": 1,
                    "items": ["Widget", "Gadget"],
                    "total": "99.99",
                }
            ]
        }
    }

    @model_validator(mode="after")
    def validate_order(self) -> "Order":
        """Validate order-level constraints."""
        if len(self.items) > 100:
            raise ValueError("Cannot have more than 100 items per order")
        return self


class AppSettings(BaseSettings):
    """Application settings loaded from environment variables.

    Pydantic Settings automatically reads from environment variables
    and .env files, with type conversion and validation.
    """

    # Database settings
    database_url: str = "sqlite:///./app.db"
    database_pool_size: int = Field(default=5, ge=1, le=100)

    # API settings
    api_key: SecretStr  # Hides value in logs/repr
    debug: bool = False
    log_level: str = Field(default="INFO")

    model_config = {
        "env_file": ".env",
        "env_file_encoding": "utf-8",
        "env_prefix": "APP_",  # APP_DATABASE_URL, APP_API_KEY, etc.
    }


if __name__ == "__main__":
    # Valid user creation
    user = User(
        id=1,
        name="Alice Smith",
        email="alice@example.com",
        age=30,
    )
    print(f"User: {user.model_dump_json(indent=2)}")

    # Type coercion - string "42" becomes int 42
    user2 = User(
        id="42",  # type: ignore - Pydantic converts this
        name="Bob",
        email="bob@example.com",
        age="25",  # type: ignore - Also converted
    )
    print(f"User2 ID type: {type(user2.id)}")  # <class 'int'>

    # Validation error example
    try:
        invalid_user = User(
            id=1,
            name="",
            email="invalid-email",
            age=200,
        )
    except Exception as e:
        print(f"\nValidation error: {e}")

    # Order with nested validation
    order = Order(
        order_id="ORD-001",
        user_id=1,
        items=["Laptop", "Mouse"],
        total=Decimal("1299.99"),
    )
    print(f"\nOrder: {order.model_dump_json(indent=2)}")

Key points:

• Pydantic validates and coerces types at runtime
• Use Field() for constraints like min_length, ge, pattern
• SecretStr hides sensitive values from logs
• BaseSettings loads config from environment variables automatically

---

Common Patterns

Pattern: Function Overloads

from typing import overload

@overload
def process(value: str) -> str: ...
@overload
def process(value: int) -> int: ...

def process(value: str | int) -> str | int:
    if isinstance(value, str):
        return value.upper()
    return value * 2

Pattern: TypedDict for JSON-like Data

from typing import TypedDict, Required, NotRequired

class UserDict(TypedDict):
    id: Required[int]
    name: Required[str]
    email: NotRequired[str]

user: UserDict = {"id": 1, "name": "Alice"}

Pattern: Callable Types

from collections.abc import Callable

def apply_twice(func: Callable[[int], int], value: int) -> int:
    return func(func(value))

result = apply_twice(lambda x: x * 2, 5)  # 20

---

Pitfalls to Avoid

Don't do this:

# Using Any defeats the purpose of type hints
from typing import Any

def process(data: Any) -> Any:
    return data.do_something()

Do this instead:

# Use specific types or protocols
from typing import Protocol

class Processable(Protocol):
    def do_something(self) -> str: ...

def process(data: Processable) -> str:
    return data.do_something()

---

Don't do this:

# Mutable default arguments with type hints
def add_item(item: str, items: list[str] = []) -> list[str]:
    items.append(item)
    return items

Do this instead:

# Use None and create new list in function body
def add_item(item: str, items: list[str] | None = None) -> list[str]:
    if items is None:
        items = []
    items.append(item)
    return items

---

See Also

• pydantic-validation.md - Deep dive into Pydantic models
• error-handling.md - Typing exception handling
• pytest-essentials.md - Type hints in tests

Docker Deployment for Python

Contents

• Quick Snippets
• Core Concepts
• Production Examples
• Example 1: Basic Flask/FastAPI Dockerfile
• Example 2: Multi-Stage Build
• Example 3: Docker Compose for Development
• Example 4: Production Kubernetes Deployment
• Common Patterns
• Pitfalls to Avoid
• See Also

---

Quick Snippets

---

Core Concepts

Docker containerization solves the "works on my machine" problem:

• Consistency: Same environment from dev to production
• Isolation: No dependency conflicts between applications
• Portability: Run anywhere Docker is installed
• Reproducibility: Versioned images for reliable deployments

Base Image Options:

Recommendation: Use slim for most apps—good balance of size and compatibility.

---

Production Examples

Example 1: Basic Flask/FastAPI Dockerfile

Use case: Simple web application containerization.

# Dockerfile
FROM python:3.12-slim

# Prevent Python from writing .pyc files and buffering stdout/stderr
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# Set working directory
WORKDIR /app

# Install system dependencies (if needed)
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements first for better layer caching
COPY requirements.txt .

# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Create non-root user for security
RUN useradd --create-home --shell /bin/bash appuser
RUN chown -R appuser:appuser /app
USER appuser

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

# Run with production server
CMD ["gunicorn", "app:app", "--bind", "0.0.0.0:8000", "--workers", "4"]

.dockerignore:

# Version control
.git
.gitignore

# Python
__pycache__
*.py[cod]
*$py.class
*.so
.Python
venv/
env/
.venv/
.env

# Testing
tests/
pytest_cache/
.coverage
htmlcov/

# IDE
.idea/
.vscode/
*.swp
*.swo

# Docker
Dockerfile
docker-compose*.yml
.dockerignore

# Documentation
docs/
*.md
!README.md

# Local config
.env.local
.env.development
*.log

Build and run:

# Build the image
docker build -t myapp:latest .

# Run the container
docker run -d \
    --name myapp \
    -p 8000:8000 \
    -e DATABASE_URL="postgresql://user:pass@db:5432/mydb" \
    myapp:latest

# View logs
docker logs -f myapp

# Stop and remove
docker stop myapp && docker rm myapp

Key points:

• Layer ordering matters—copy requirements.txt first for caching
• Use non-root user for security
• PYTHONUNBUFFERED=1 ensures logs are visible immediately
• Health checks enable orchestrator monitoring

---

Example 2: Multi-Stage Build

Use case: Minimize image size by separating build and runtime.

# Stage 1: Build stage
FROM python:3.12-slim AS builder

WORKDIR /app

# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Create virtual environment
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt


# Stage 2: Production stage
FROM python:3.12-slim AS production

WORKDIR /app

# Install runtime dependencies only
RUN apt-get update && apt-get install -y --no-install-recommends \
    libpq5 \
    && rm -rf /var/lib/apt/lists/*

# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Copy application code
COPY src/ ./src/
COPY alembic/ ./alembic/
COPY alembic.ini .

# Create non-root user
RUN useradd --create-home --shell /bin/bash appuser \
    && chown -R appuser:appuser /app
USER appuser

# Environment
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

CMD ["gunicorn", "src.main:app", "--bind", "0.0.0.0:8000", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker"]

With Poetry:

# Multi-stage build with Poetry
FROM python:3.12-slim AS builder

WORKDIR /app

# Install Poetry
RUN pip install poetry==1.7.1

# Copy dependency files
COPY pyproject.toml poetry.lock ./

# Export to requirements.txt (no dev dependencies)
RUN poetry export -f requirements.txt --without-hashes --only main > requirements.txt

# Create virtual environment and install
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
RUN pip install --no-cache-dir -r requirements.txt


FROM python:3.12-slim AS production

WORKDIR /app

# Copy virtual environment
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Copy application
COPY src/ ./src/

# Non-root user
RUN useradd --create-home appuser && chown -R appuser:appuser /app
USER appuser

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

EXPOSE 8000
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

Size comparison:

# Single stage (full image)
myapp:single    1.2GB

# Multi-stage (slim runtime)
myapp:multi     180MB

Key points:

• Builder stage has compilers and dev tools
• Production stage only has runtime dependencies
• Virtual environment copied intact between stages
• Final image is much smaller and more secure

---

Example 3: Docker Compose for Development

Use case: Local development with database, cache, and hot reload.

# docker-compose.yml
services:
  app:
    build:
      context: .
      dockerfile: Dockerfile.dev
    ports:
      - "8000:8000"
    volumes:
      - ./src:/app/src:cached
      - ./tests:/app/tests:cached
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/myapp
      - REDIS_URL=redis://redis:6379/0
      - DEBUG=true
      - LOG_LEVEL=DEBUG
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    command: uvicorn src.main:app --host 0.0.0.0 --port 8000 --reload

  db:
    image: postgres:16-alpine
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=myapp
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./scripts/init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  # Database admin UI
  adminer:
    image: adminer
    ports:
      - "8080:8080"
    depends_on:
      - db

volumes:
  postgres_data:
  redis_data:

Development Dockerfile (Dockerfile.dev):

FROM python:3.12-slim

WORKDIR /app

# Install dev dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Install dependencies
COPY requirements.txt requirements-dev.txt ./
RUN pip install --no-cache-dir -r requirements.txt -r requirements-dev.txt

# Copy application
COPY . .

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

EXPOSE 8000

# Development server with hot reload
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

Development workflow:

# Start all services
docker compose up -d

# View logs
docker compose logs -f app

# Run tests in container
docker compose exec app pytest

# Run migrations
docker compose exec app alembic upgrade head

# Access database shell
docker compose exec db psql -U postgres -d myapp

# Rebuild after dependency changes
docker compose build app
docker compose up -d app

# Stop everything
docker compose down

# Stop and remove volumes (reset data)
docker compose down -v

Production compose (docker-compose.prod.yml):

# docker-compose.prod.yml
services:
  app:
    image: myregistry.com/myapp:${VERSION:-latest}
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=${DATABASE_URL}
      - REDIS_URL=${REDIS_URL}
      - SECRET_KEY=${SECRET_KEY}
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: '1'
          memory: 512M
        reservations:
          cpus: '0.5'
          memory: 256M
      restart_policy:
        condition: on-failure
        delay: 5s
        max_attempts: 3
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 40s

Key points:

• Use volumes to mount source code for hot reload
• depends_on with health checks ensures startup order
• Separate dev and prod compose files
• Use environment variables for configuration

---

Example 4: Production Kubernetes Deployment

Use case: Deploy to Kubernetes with proper health checks and scaling.

Deployment manifest (k8s/deployment.yaml):

apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
  labels:
    app: myapp
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
  template:
    metadata:
      labels:
        app: myapp
    spec:
      # Non-root security context
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        fsGroup: 1000

      containers:
        - name: myapp
          image: myregistry.com/myapp:1.0.0
          ports:
            - containerPort: 8000
              name: http

          # Environment from ConfigMap and Secret
          envFrom:
            - configMapRef:
                name: myapp-config
            - secretRef:
                name: myapp-secrets

          # Resource limits
          resources:
            requests:
              cpu: 250m
              memory: 256Mi
            limits:
              cpu: 1000m
              memory: 512Mi

          # Probes
          livenessProbe:
            httpGet:
              path: /health
              port: http
            initialDelaySeconds: 10
            periodSeconds: 30
            timeoutSeconds: 5
            failureThreshold: 3

          readinessProbe:
            httpGet:
              path: /ready
              port: http
            initialDelaySeconds: 5
            periodSeconds: 10
            timeoutSeconds: 3
            failureThreshold: 3

          startupProbe:
            httpGet:
              path: /health
              port: http
            initialDelaySeconds: 5
            periodSeconds: 5
            failureThreshold: 30  # 5 * 30 = 150s max startup time

          # Security settings
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop:
                - ALL

          # Mount for tmp files if needed
          volumeMounts:
            - name: tmp
              mountPath: /tmp

      volumes:
        - name: tmp
          emptyDir: {}

      # Pod anti-affinity for high availability
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
            - weight: 100
              podAffinityTerm:
                labelSelector:
                  matchLabels:
                    app: myapp
                topologyKey: kubernetes.io/hostname

Service and Ingress:

# k8s/service.yaml
apiVersion: v1
kind: Service
metadata:
  name: myapp
spec:
  type: ClusterIP
  ports:
    - port: 80
      targetPort: 8000
      name: http
  selector:
    app: myapp
---
# k8s/ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: myapp
  annotations:
    kubernetes.io/ingress.class: nginx
    cert-manager.io/cluster-issuer: letsencrypt-prod
spec:
  tls:
    - hosts:
        - api.example.com
      secretName: myapp-tls
  rules:
    - host: api.example.com
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: myapp
                port:
                  number: 80

ConfigMap and Secrets:

# k8s/config.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: myapp-config
data:
  LOG_LEVEL: "INFO"
  WORKERS: "4"
  ENVIRONMENT: "production"
---
apiVersion: v1
kind: Secret
metadata:
  name: myapp-secrets
type: Opaque
stringData:
  DATABASE_URL: "postgresql://user:password@db:5432/myapp"
  SECRET_KEY: "your-secret-key-here"

Health check endpoints in FastAPI:

# src/health.py
from fastapi import APIRouter, Response
from sqlalchemy import text

router = APIRouter(tags=["health"])


@router.get("/health")
async def health():
    """Liveness probe - is the app alive?"""
    return {"status": "healthy"}


@router.get("/ready")
async def ready(db: AsyncSession = Depends(get_db)):
    """Readiness probe - can the app serve traffic?"""
    try:
        # Check database connection
        await db.execute(text("SELECT 1"))
        return {"status": "ready", "checks": {"database": "ok"}}
    except Exception as e:
        return Response(
            content=f'{{"status": "not ready", "error": "{str(e)}"}}',
            status_code=503,
            media_type="application/json",
        )

Key points:

• Use all three probe types: liveness, readiness, startup
• Set appropriate resource requests and limits
• Run as non-root with read-only filesystem
• Use pod anti-affinity for high availability
• Separate ConfigMaps (non-sensitive) from Secrets (sensitive)

---

Common Patterns

Pattern: Graceful Shutdown

# main.py
import signal
import asyncio

shutdown_event = asyncio.Event()

def handle_sigterm(*args):
    shutdown_event.set()

signal.signal(signal.SIGTERM, handle_sigterm)

@app.on_event("shutdown")
async def shutdown():
    # Finish in-flight requests
    await asyncio.sleep(5)

Pattern: Build Args for Versioning

ARG VERSION=unknown
ARG BUILD_DATE=unknown
ARG GIT_SHA=unknown

LABEL version=$VERSION
LABEL build-date=$BUILD_DATE
LABEL git-sha=$GIT_SHA

ENV APP_VERSION=$VERSION

docker build \
    --build-arg VERSION=1.0.0 \
    --build-arg BUILD_DATE=$(date -u +%Y-%m-%dT%H:%M:%SZ) \
    --build-arg GIT_SHA=$(git rev-parse HEAD) \
    -t myapp:1.0.0 .

Pattern: Multi-Platform Builds

# Build for multiple architectures
docker buildx build \
    --platform linux/amd64,linux/arm64 \
    -t myapp:latest \
    --push .

---

Pitfalls to Avoid

Don't do this:

# Running as root
FROM python:3.12-slim
COPY . /app
CMD ["python", "app.py"]
# Runs as root - security risk!

Do this instead:

FROM python:3.12-slim
RUN useradd --create-home appuser
WORKDIR /app
COPY --chown=appuser:appuser . .
USER appuser
CMD ["python", "app.py"]

---

Don't do this:

# Poor layer caching
COPY . .
RUN pip install -r requirements.txt
# Every code change rebuilds dependencies!

Do this instead:

# Requirements first, code second
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
# Code changes don't rebuild deps

---

Don't do this:

# Using latest tag
FROM python:latest
# Unpredictable, breaks builds randomly

Do this instead:

# Pin exact version
FROM python:3.12.2-slim
# Reproducible, predictable builds

---

See Also

• poetry-workflow.md - Managing dependencies with Poetry
• pyproject-config.md - Project configuration
• async-programming.md - Async patterns for web apps

Poetry Workflow and Dependency Management

Contents

• Quick Snippets
• Core Concepts
• Production Examples
• Example 1: Project Setup and Configuration
• Example 2: Managing Dependencies
• Example 3: Dependency Groups and Environments
• Example 4: Publishing to PyPI
• Common Patterns
• Pitfalls to Avoid
• See Also

---

Quick Snippets

---

Core Concepts

Poetry is a modern Python dependency management and packaging tool that:

• Unified Workflow: Handles environment creation, dependency management, and publishing
• Declarative Configuration: All config in pyproject.toml (PEP 517/518 compliant)
• Deterministic Builds: Lock file ensures identical dependencies everywhere
• SAT Solver: Sophisticated dependency resolution avoids version conflicts

Poetry vs Traditional Tools:

---

Production Examples

Example 1: Project Setup and Configuration

Use case: Create a new project with proper structure and configuration.

# Install Poetry using pipx (recommended)
pipx install poetry

# Verify installation
poetry --version

# Create new project with standard layout
poetry new my-service
cd my-service

# Or initialize in existing directory
mkdir existing-project && cd existing-project
poetry init --no-interaction

# Project structure created:
# my-service/
# ├── pyproject.toml
# ├── README.md
# ├── my_service/
# │   └── __init__.py
# └── tests/
#     └── __init__.py

pyproject.toml (generated and customized):

[tool.poetry]
name = "my-service"
version = "0.1.0"
description = "A production-ready Python service"
authors = ["Your Name <your.email@example.com>"]
license = "MIT"
readme = "README.md"
homepage = "https://github.com/yourorg/my-service"
repository = "https://github.com/yourorg/my-service"
documentation = "https://my-service.readthedocs.io"
keywords = ["service", "api", "python"]
classifiers = [
    "Development Status :: 4 - Beta",
    "Intended Audience :: Developers",
    "Operating System :: OS Independent",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
]

# Package discovery
packages = [{include = "my_service", from = "src"}]

# Python version constraint
[tool.poetry.dependencies]
python = "^3.11"

# Build system (required for PEP 517)
[build-system]
requires = ["poetry-core>=1.5.0"]
build-backend = "poetry.core.masonry.api"

Key configuration options:

# Use src layout (recommended for larger projects)
packages = [{include = "my_service", from = "src"}]

# Include additional files
include = ["CHANGELOG.md", "data/*.json"]

# Exclude files from package
exclude = ["tests/*", "docs/*"]

# CLI entry points
[tool.poetry.scripts]
my-cli = "my_service.cli:main"

# Plugin entry points
[tool.poetry.plugins."my_service.plugins"]
default = "my_service.plugins.default:DefaultPlugin"

Key points:

• Use pipx to install Poetry globally (avoids dependency conflicts)
• pyproject.toml replaces setup.py, setup.cfg, and requirements.txt
• Version constraints use semantic versioning: ^3.11 means >=3.11.0 <4.0.0
• The [build-system] section is required for PEP 517 compliance

---

Example 2: Managing Dependencies

Use case: Add, update, and manage project dependencies.

# Add runtime dependencies
poetry add fastapi
poetry add "uvicorn[standard]"  # With extras
poetry add pydantic==2.5.0  # Specific version
poetry add "sqlalchemy>=2.0,<3.0"  # Version range

# Add dependencies from git
poetry add git+https://github.com/org/repo.git
poetry add git+https://github.com/org/repo.git#branch=develop
poetry add git+https://github.com/org/repo.git#tag=v1.0.0

# Add from local path (for development)
poetry add ../my-local-package --editable

# Update dependencies
poetry update  # Update all
poetry update fastapi pydantic  # Update specific
poetry update --dry-run  # Preview changes

# Show dependency tree
poetry show --tree

# Show outdated packages
poetry show --outdated

# Remove dependencies
poetry remove fastapi

# Lock without installing
poetry lock

# Install from lock file only (production)
poetry install --no-root --only main

Lock file explained (poetry.lock):

# This file is auto-generated by Poetry
# DO NOT edit manually

[[package]]
name = "fastapi"
version = "0.109.0"
description = "FastAPI framework"
optional = false
python-versions = ">=3.8"

[package.dependencies]
pydantic = ">=1.7.4,<2.0.0 || >2.0.0,<3.0.0"
starlette = ">=0.35.0,<0.36.0"
typing-extensions = ">=4.8.0"

[package.extras]
all = ["email-validator", "httpx", "python-multipart", ...]

[[package]]
name = "starlette"
version = "0.35.1"
# ... transitive dependencies are also locked

Environment management:

# Poetry creates virtual env automatically
# Default location: {cache-dir}/virtualenvs/

# Configure to create .venv in project directory
poetry config virtualenvs.in-project true

# Show current environment info
poetry env info

# List all environments
poetry env list

# Remove environment
poetry env remove python3.11

# Use specific Python version
poetry env use python3.12
poetry env use /usr/bin/python3.12

Key points:

• The lock file pins exact versions of ALL dependencies (including transitive)
• Always commit poetry.lock to version control
• Use poetry update carefully—it can upgrade many packages
• --no-root skips installing the current project (useful in Docker)

---

Example 3: Dependency Groups and Environments

Use case: Organize dependencies for different environments (dev, test, prod).

# pyproject.toml

[tool.poetry.dependencies]
python = "^3.11"
# Production dependencies
fastapi = "^0.109.0"
uvicorn = {extras = ["standard"], version = "^0.27.0"}
pydantic = "^2.5.0"
sqlalchemy = "^2.0.0"
alembic = "^1.13.0"

# Optional dependencies (extras)
[tool.poetry.extras]
postgres = ["psycopg2-binary"]
mysql = ["pymysql"]
all = ["psycopg2-binary", "pymysql"]

# Development tools
[tool.poetry.group.dev.dependencies]
black = "^24.1.0"
ruff = "^0.1.14"
mypy = "^1.8.0"
pre-commit = "^3.6.0"
ipython = "^8.20.0"

# Testing
[tool.poetry.group.test.dependencies]
pytest = "^8.0.0"
pytest-cov = "^4.1.0"
pytest-asyncio = "^0.23.0"
httpx = "^0.26.0"  # For testing FastAPI
factory-boy = "^3.3.0"

# Documentation
[tool.poetry.group.docs]
optional = true  # Not installed by default

[tool.poetry.group.docs.dependencies]
mkdocs = "^1.5.0"
mkdocs-material = "^9.5.0"
mkdocstrings = {extras = ["python"], version = "^0.24.0"}

Working with dependency groups:

# Install all groups (default)
poetry install

# Install only main + specific groups
poetry install --only main,test

# Skip specific groups
poetry install --without docs,dev

# Install optional group
poetry install --with docs

# Export specific groups
poetry export -f requirements.txt --only main -o requirements.txt
poetry export -f requirements.txt --with dev,test -o requirements-dev.txt

# Production deployment (main deps only)
poetry install --no-root --only main

# CI testing (main + test deps)
poetry install --only main,test

Environment-specific installation script:

#!/bin/bash
# install.sh - Install dependencies based on environment

set -e

case "${ENVIRONMENT:-development}" in
    production)
        echo "Installing production dependencies..."
        poetry install --no-root --only main
        ;;
    test)
        echo "Installing test dependencies..."
        poetry install --only main,test
        ;;
    development)
        echo "Installing all development dependencies..."
        poetry install
        ;;
    *)
        echo "Unknown environment: $ENVIRONMENT"
        exit 1
        ;;
esac

GitHub Actions CI example:

# .github/workflows/ci.yml
name: CI

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install Poetry
        uses: snok/install-poetry@v1
        with:
          virtualenvs-create: true
          virtualenvs-in-project: true

      - name: Load cached venv
        id: cached-poetry-dependencies
        uses: actions/cache@v4
        with:
          path: .venv
          key: venv-${{ runner.os }}-${{ hashFiles('**/poetry.lock') }}

      - name: Install dependencies
        if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
        run: poetry install --only main,test

      - name: Run tests
        run: poetry run pytest --cov=my_service --cov-report=xml

      - name: Type check
        run: poetry run mypy my_service

Key points:

• Use dependency groups to organize by purpose (dev, test, docs)
• Mark groups as optional = true if not needed by default
• Export to requirements.txt for Docker builds or compatibility
• Cache the virtual environment in CI for faster builds

---

Example 4: Publishing to PyPI

Use case: Build and publish packages to PyPI or private registries.

# Build distributions
poetry build
# Creates:
# dist/
# ├── my_service-0.1.0-py3-none-any.whl
# └── my_service-0.1.0.tar.gz

# Check the built package
poetry run twine check dist/*

# Publish to PyPI
poetry publish

# Publish to TestPyPI first
poetry config repositories.testpypi https://test.pypi.org/legacy/
poetry publish -r testpypi

# Publish with token (non-interactive)
poetry config pypi-token.pypi your-token-here
poetry publish --no-interaction

# Build and publish in one command
poetry publish --build

Configuring private registries:

# Add private registry
poetry config repositories.private https://pypi.mycompany.com/simple/

# Set credentials
poetry config http-basic.private username password

# Or use token
poetry config pypi-token.private your-token-here

# Publish to private registry
poetry publish -r private

# Install from private registry
poetry source add private https://pypi.mycompany.com/simple/
poetry add my-private-package --source private

pyproject.toml with sources:

# Configure package sources
[[tool.poetry.source]]
name = "private"
url = "https://pypi.mycompany.com/simple/"
priority = "supplemental"  # Use for specific packages only

[[tool.poetry.source]]
name = "PyPI"
priority = "primary"

Version management:

# Bump version (follows semver)
poetry version patch   # 0.1.0 → 0.1.1
poetry version minor   # 0.1.0 → 0.2.0
poetry version major   # 0.1.0 → 1.0.0
poetry version prepatch  # 0.1.0 → 0.1.1a0
poetry version prerelease  # 0.1.0 → 0.1.0a0
poetry version 2.0.0  # Set exact version

# Show current version
poetry version --short

Automated release workflow:

# .github/workflows/release.yml
name: Release

on:
  push:
    tags:
      - 'v*'

jobs:
  release:
    runs-on: ubuntu-latest
    permissions:
      id-token: write  # For trusted publishing

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install Poetry
        uses: snok/install-poetry@v1

      - name: Build
        run: poetry build

      - name: Publish to PyPI
        uses: pypa/gh-action-pypi-publish@release/v1
        # Uses trusted publishing - no token needed!

Key points:

• Always test with TestPyPI before publishing to production PyPI
• Use trusted publishing (OIDC) in GitHub Actions—no tokens needed
• Semantic versioning: major.minor.patch (breaking.feature.fix)
• Keep poetry.lock in version control for reproducible builds

---

Common Patterns

Pattern: Monorepo with Multiple Packages

# packages/core/pyproject.toml
[tool.poetry]
name = "myorg-core"

[tool.poetry.dependencies]
python = "^3.11"

# packages/api/pyproject.toml
[tool.poetry]
name = "myorg-api"

[tool.poetry.dependencies]
python = "^3.11"
myorg-core = {path = "../core", develop = true}

Pattern: Platform-Specific Dependencies

[tool.poetry.dependencies]
pywin32 = {version = "^306", markers = "sys_platform == 'win32'"}
uvloop = {version = "^0.19", markers = "sys_platform != 'win32'"}

Pattern: Dynamic Versioning

# Use poetry-dynamic-versioning plugin
[tool.poetry-dynamic-versioning]
enable = true
vcs = "git"
style = "pep440"

---

Pitfalls to Avoid

Don't do this:

# Installing Poetry with pip in your project
pip install poetry  # Creates dependency conflicts!

Do this instead:

# Install Poetry system-wide with pipx
pipx install poetry

---

Don't do this:

# Ignoring the lock file
poetry install --no-lock  # Non-deterministic builds!

Do this instead:

# Always use lock file
poetry install  # Uses poetry.lock for exact versions
poetry lock  # Update lock file when needed

---

Don't do this:

# Overly permissive version constraints
[tool.poetry.dependencies]
requests = "*"  # Any version - dangerous!

Do this instead:

# Specific constraints with caret
[tool.poetry.dependencies]
requests = "^2.31.0"  # >=2.31.0 <3.0.0

---

See Also

• pyproject-config.md - pyproject.toml configuration
• docker-deployment.md - Containerizing Python apps
• project-structure.md - Project layout

pyproject.toml Configuration

Contents

• Quick Snippets
• Core Concepts
• Production Examples
• Example 1: Minimal Configuration
• Example 2: Full Project Configuration
• Example 3: Tool Configuration
• Example 4: Build System Configuration
• Common Patterns
• Pitfalls to Avoid
• See Also

---

Quick Snippets

---

Core Concepts

pyproject.toml is the standard Python project configuration file (PEP 518, 621):

• Single Source of Truth: Replaces setup.py, setup.cfg, requirements.txt
• Declarative: Describes what, not how (unlike executable setup.py)
• Tool-Agnostic: Works with pip, Poetry, Hatch, PDM, and more
• Standardized Metadata: PEP 621 defines project metadata format

Key PEPs:

---

Production Examples

Example 1: Minimal Configuration

Use case: Simple package with basic metadata.

# Minimal pyproject.toml for a simple package

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "my-package"
version = "0.1.0"
description = "A simple Python package"
requires-python = ">=3.9"
dependencies = [
    "requests>=2.28.0",
]

Build and install:

# Build distributions
python -m build

# Install in development mode
pip install -e .

# Install from source
pip install .

Key points:

• [build-system] is required for modern builds
• requires-python prevents installation on incompatible versions
• Dependencies use PEP 508 format (version specifiers, markers)

---

Example 2: Full Project Configuration

Use case: Production-ready package with complete metadata.

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "my-awesome-package"
version = "1.0.0"
description = "A production-ready Python package"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10"
authors = [
    {name = "Your Name", email = "your.email@example.com"},
]
maintainers = [
    {name = "Maintainer Name", email = "maintainer@example.com"},
]
keywords = ["python", "api", "automation"]
classifiers = [
    "Development Status :: 5 - Production/Stable",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: MIT License",
    "Operating System :: OS Independent",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Topic :: Software Development :: Libraries :: Python Modules",
    "Typing :: Typed",
]

# Runtime dependencies
dependencies = [
    "httpx>=0.25.0",
    "pydantic>=2.0.0,<3.0.0",
    "click>=8.0.0",
]

# Optional dependencies (extras)
[project.optional-dependencies]
dev = [
    "pytest>=7.0.0",
    "pytest-cov>=4.0.0",
    "black>=23.0.0",
    "ruff>=0.1.0",
    "mypy>=1.0.0",
]
docs = [
    "mkdocs>=1.5.0",
    "mkdocs-material>=9.0.0",
]
all = [
    "my-awesome-package[dev,docs]",
]

# Project URLs
[project.urls]
Homepage = "https://github.com/yourorg/my-awesome-package"
Documentation = "https://my-awesome-package.readthedocs.io"
Repository = "https://github.com/yourorg/my-awesome-package.git"
Changelog = "https://github.com/yourorg/my-awesome-package/blob/main/CHANGELOG.md"
"Bug Tracker" = "https://github.com/yourorg/my-awesome-package/issues"

# CLI entry points
[project.scripts]
my-cli = "my_awesome_package.cli:main"
my-tool = "my_awesome_package.tools:run"

# GUI entry points (Windows)
[project.gui-scripts]
my-gui = "my_awesome_package.gui:main"

# Plugin entry points
[project.entry-points."my_awesome_package.plugins"]
default = "my_awesome_package.plugins.default:DefaultPlugin"
advanced = "my_awesome_package.plugins.advanced:AdvancedPlugin"

Installing with extras:

# Install with development dependencies
pip install -e ".[dev]"

# Install with documentation dependencies
pip install -e ".[docs]"

# Install all extras
pip install -e ".[all]"

# Install specific extras for production
pip install "my-awesome-package[postgres,redis]"

Key points:

• Use classifiers for PyPI categorization and discoverability
• [project.urls] appears on PyPI package page
• Entry points enable CLI tools and plugin systems
• Optional dependencies allow flexible installation

---

Example 3: Tool Configuration

Use case: Configure development tools in pyproject.toml.

# ============================================================
# TOOL CONFIGURATIONS
# ============================================================

# Black - Code formatter
[tool.black]
line-length = 88
target-version = ["py310", "py311", "py312"]
include = '\.pyi?$'
exclude = '''
/(
    \.git
    | \.hg
    | \.mypy_cache
    | \.tox
    | \.venv
    | _build
    | buck-out
    | build
    | dist
)/
'''

# Ruff - Fast linter (replaces flake8, isort, etc.)
[tool.ruff]
line-length = 88
target-version = "py310"
select = [
    "E",    # pycodestyle errors
    "W",    # pycodestyle warnings
    "F",    # Pyflakes
    "I",    # isort
    "B",    # flake8-bugbear
    "C4",   # flake8-comprehensions
    "UP",   # pyupgrade
    "ARG",  # flake8-unused-arguments
    "SIM",  # flake8-simplify
]
ignore = [
    "E501",  # line too long (handled by black)
    "B008",  # function call in argument defaults
]

[tool.ruff.per-file-ignores]
"__init__.py" = ["F401"]  # Allow unused imports
"tests/*" = ["ARG001"]    # Allow unused arguments in tests

[tool.ruff.isort]
known-first-party = ["my_awesome_package"]
force-single-line = true

# MyPy - Type checker
[tool.mypy]
python_version = "3.11"
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true

[[tool.mypy.overrides]]
module = "tests.*"
disallow_untyped_defs = false

[[tool.mypy.overrides]]
module = [
    "httpx.*",
    "pytest.*",
]
ignore_missing_imports = true

# Pytest - Testing
[tool.pytest.ini_options]
minversion = "7.0"
testpaths = ["tests"]
pythonpath = ["src"]
addopts = [
    "-ra",
    "-q",
    "--strict-markers",
    "--strict-config",
]
markers = [
    "slow: marks tests as slow (deselect with '-m \"not slow\"')",
    "integration: marks tests as integration tests",
]
filterwarnings = [
    "error",
    "ignore::DeprecationWarning",
]
asyncio_mode = "auto"

# Coverage
[tool.coverage.run]
source = ["src"]
branch = true
parallel = true
omit = [
    "*/__init__.py",
    "*/tests/*",
]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "raise AssertionError",
    "raise NotImplementedError",
    "if __name__ == .__main__.:",
    "if TYPE_CHECKING:",
]
fail_under = 80
show_missing = true

[tool.coverage.paths]
source = [
    "src/",
    "*/site-packages/",
]

# Setuptools package discovery
[tool.setuptools.packages.find]
where = ["src"]

[tool.setuptools.package-data]
"*" = ["py.typed", "*.pyi"]

Running tools:

# Format code
black src/ tests/
ruff format src/ tests/

# Lint code
ruff check src/ tests/
ruff check --fix src/ tests/  # Auto-fix

# Type check
mypy src/

# Run tests with coverage
pytest --cov=my_awesome_package --cov-report=html

# All checks (CI)
black --check src/ tests/
ruff check src/ tests/
mypy src/
pytest --cov=my_awesome_package --cov-fail-under=80

Key points:

• [tool.*] sections are tool-specific (not standardized)
• Ruff is modern and fast—can replace flake8, isort, pyupgrade
• Use strict mypy settings for new projects
• Configure pytest markers for test categorization

---

Example 4: Build System Configuration

Use case: Configure different build backends.

Setuptools (traditional):

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[tool.setuptools]
package-dir = {"" = "src"}
include-package-data = true

[tool.setuptools.packages.find]
where = ["src"]
include = ["my_package*"]
exclude = ["tests*"]

[tool.setuptools.package-data]
my_package = ["py.typed", "data/*.json"]

[tool.setuptools.dynamic]
version = {attr = "my_package.__version__"}
readme = {file = ["README.md"]}

Hatchling (modern, fast):

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.version]
path = "src/my_package/__init__.py"

[tool.hatch.build.targets.sdist]
include = [
    "/src",
    "/tests",
]

[tool.hatch.build.targets.wheel]
packages = ["src/my_package"]

Flit (simple, pure Python):

[build-system]
requires = ["flit_core>=3.4"]
build-backend = "flit_core.buildapi"

[tool.flit.module]
name = "my_package"

[tool.flit.sdist]
include = ["doc/"]
exclude = ["doc/*.html"]

Poetry-core (Poetry ecosystem):

[build-system]
requires = ["poetry-core>=1.5.0"]
build-backend = "poetry.core.masonry.api"

# Poetry uses [tool.poetry] instead of [project]
[tool.poetry]
name = "my-package"
version = "1.0.0"
description = "My package"
authors = ["Your Name <you@example.com>"]
packages = [{include = "my_package", from = "src"}]

Maturin (Rust extensions):

[build-system]
requires = ["maturin>=1.0"]
build-backend = "maturin"

[tool.maturin]
features = ["pyo3/extension-module"]
python-source = "python"
module-name = "my_rust_package._core"

Build and verify:

# Build with any backend
python -m build

# Check distribution
twine check dist/*

# Inspect wheel contents
unzip -l dist/*.whl

# Install and test
pip install dist/*.whl
python -c "import my_package; print(my_package.__version__)"

Key points:

• Choose build backend based on needs (setuptools for C extensions, hatch for speed)
• Setuptools is most compatible, Hatch/Flit are simpler for pure Python
• Poetry-core only for Poetry-managed projects
• Maturin for Rust-based Python extensions

---

Common Patterns

Pattern: Dynamic Version from Git

[build-system]
requires = ["setuptools>=61.0", "setuptools-scm>=8.0"]
build-backend = "setuptools.build_meta"

[tool.setuptools_scm]
write_to = "src/my_package/_version.py"
version_scheme = "python-simplified-semver"

Pattern: Conditional Dependencies

[project]
dependencies = [
    "tomli>=2.0; python_version < '3.11'",
    "typing-extensions>=4.0; python_version < '3.10'",
]

Pattern: Platform-Specific Dependencies

[project]
dependencies = [
    "pywin32>=306; sys_platform == 'win32'",
    "uvloop>=0.19; sys_platform != 'win32'",
]

Pattern: Namespace Packages

[tool.setuptools.packages.find]
where = ["src"]
namespaces = true

---

Pitfalls to Avoid

Don't do this:

# Missing build-system section
[project]
name = "my-package"
# pip will use legacy setup.py behavior!

Do this instead:

# Always include build-system
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "my-package"

---

Don't do this:

# Mixing Poetry's [tool.poetry] with [project]
[project]
name = "my-package"

[tool.poetry.dependencies]
python = "^3.11"
# Confusing! Which one is used?

Do this instead:

# Use one or the other consistently
# For Poetry:
[tool.poetry]
name = "my-package"
# OR for standard PEP 621:
[project]
name = "my-package"

---

Don't do this:

# Overly broad version constraints
[project]
dependencies = [
    "requests",  # Any version!
]

Do this instead:

# Specify minimum versions
[project]
dependencies = [
    "requests>=2.28.0",
    "pydantic>=2.0.0,<3.0.0",
]

---

See Also

• poetry-workflow.md - Poetry dependency management
• project-structure.md - Project layout
• code-quality.md - Linting and formatting

Asynchronous Programming and Concurrency

Contents

• Quick Snippets
• Core Concepts
• Production Examples
• Example 1: TaskGroup for Structured Concurrency
• Example 2: Timeout Handling
• Example 3: Async HTTP Client with Retry
• Example 4: Thread Pool for Blocking I/O
• Common Patterns
• Pitfalls to Avoid
• See Also

---

Quick Snippets

---

Core Concepts

Asyncio enables concurrent I/O-bound operations without threads. Key concepts:

• Coroutines: Functions defined with async def that can be paused and resumed
• Event Loop: Manages and schedules coroutine execution
• Tasks: Wrap coroutines for concurrent execution
• Structured Concurrency: Python 3.11+ TaskGroup ensures all tasks complete or fail together

Python 3.11+ introduced significant improvements:

• asyncio.TaskGroup for structured concurrency
• asyncio.timeout() context manager
• Exception groups (except*) for handling multiple exceptions

---

Production Examples

Example 1: TaskGroup for Structured Concurrency

Use case: Fetch multiple URLs concurrently with proper error handling.

#!/usr/bin/env python3
"""Structured concurrency with TaskGroup (Python 3.11+)."""

import asyncio
from dataclasses import dataclass


@dataclass
class FetchResult:
    """Result from fetching a URL."""

    url: str
    status: int
    content_length: int


async def fetch_url(url: str) -> FetchResult:
    """Simulate fetching a URL."""
    # Simulate network delay
    await asyncio.sleep(0.1)

    # Simulate occasional failures
    if "error" in url:
        raise ValueError(f"Failed to fetch {url}")

    return FetchResult(url=url, status=200, content_length=1024)


async def fetch_all(urls: list[str]) -> list[FetchResult]:
    """Fetch all URLs concurrently using TaskGroup.

    TaskGroup ensures:
    - All tasks complete before exiting the context
    - If any task fails, all others are cancelled
    - All exceptions are collected into an ExceptionGroup
    """
    results: list[FetchResult] = []

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_url(url)) for url in urls]

    # All tasks completed successfully if we reach here
    results = [task.result() for task in tasks]
    return results


async def fetch_with_error_handling(urls: list[str]) -> list[FetchResult | None]:
    """Fetch URLs with individual error handling."""
    results: list[FetchResult | None] = []

    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_url(url)) for url in urls]
    except* ValueError as exc_group:
        # Handle ValueError exceptions from failed fetches
        for exc in exc_group.exceptions:
            print(f"Fetch error: {exc}")
        # Return results for tasks that succeeded
        results = [
            task.result() if not task.cancelled() and task.exception() is None else None
            for task in tasks
        ]
    else:
        results = [task.result() for task in tasks]

    return results


if __name__ == "__main__":
    urls = [
        "https://example.com/api/1",
        "https://example.com/api/2",
        "https://example.com/api/3",
    ]

    # Successful fetch
    results = asyncio.run(fetch_all(urls))
    for r in results:
        print(f"Fetched {r.url}: {r.status}")

    # With error handling
    urls_with_error = urls + ["https://example.com/error"]
    results = asyncio.run(fetch_with_error_handling(urls_with_error))
    print(f"Got {len([r for r in results if r])} successful results")

Key points:

• TaskGroup replaces manual gather() with exception handling
• Use except* to catch specific exceptions from the group
• All tasks are automatically cancelled if one fails (unless handled)

---

Example 2: Timeout Handling

Use case: Set timeouts for async operations to prevent hanging.

#!/usr/bin/env python3
"""Timeout handling with asyncio.timeout() (Python 3.11+)."""

import asyncio
from contextlib import suppress


async def slow_operation(duration: float) -> str:
    """Simulate a slow operation."""
    await asyncio.sleep(duration)
    return f"Completed after {duration}s"


async def fetch_with_timeout(timeout_seconds: float) -> str | None:
    """Fetch with a timeout, returning None on timeout."""
    try:
        async with asyncio.timeout(timeout_seconds):
            result = await slow_operation(2.0)
            return result
    except TimeoutError:
        print(f"Operation timed out after {timeout_seconds}s")
        return None


async def fetch_with_deadline() -> str | None:
    """Use timeout_at for absolute deadline."""
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 1.0  # 1 second from now

    try:
        async with asyncio.timeout_at(deadline):
            return await slow_operation(2.0)
    except TimeoutError:
        return None


async def multiple_with_individual_timeouts() -> list[str | None]:
    """Each task gets its own timeout."""
    async def fetch_one(url: str, timeout: float) -> str | None:
        try:
            async with asyncio.timeout(timeout):
                await asyncio.sleep(0.5)  # Simulate fetch
                return f"Fetched {url}"
        except TimeoutError:
            return None

    async with asyncio.TaskGroup() as tg:
        tasks = [
            tg.create_task(fetch_one("url1", 1.0)),
            tg.create_task(fetch_one("url2", 0.3)),  # Will timeout
            tg.create_task(fetch_one("url3", 1.0)),
        ]

    return [t.result() for t in tasks]


async def reschedule_timeout() -> str:
    """Dynamically extend timeout based on progress."""
    async with asyncio.timeout(1.0) as cm:
        await asyncio.sleep(0.5)

        # Reschedule to give more time
        cm.reschedule(asyncio.get_running_loop().time() + 2.0)

        await asyncio.sleep(1.0)  # Would have timed out without reschedule
        return "Completed with extended timeout"


if __name__ == "__main__":
    # Basic timeout
    result = asyncio.run(fetch_with_timeout(1.0))
    print(f"Result: {result}")

    # Reschedule example
    result = asyncio.run(reschedule_timeout())
    print(f"Reschedule result: {result}")

Key points:

• asyncio.timeout(seconds) replaces asyncio.wait_for()
• timeout_at() uses absolute timestamps for deadlines
• cm.reschedule() can extend timeouts dynamically

---

Example 3: Async HTTP Client with Retry

Use case: Production HTTP client with retry logic and connection pooling.

#!/usr/bin/env python3
"""Async HTTP client with retry and connection management."""

import asyncio
from dataclasses import dataclass
from typing import Any

import aiohttp
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)


@dataclass
class APIResponse:
    """Structured API response."""

    status: int
    data: dict[str, Any]
    elapsed_ms: float


class AsyncAPIClient:
    """Production async HTTP client with retry and pooling."""

    def __init__(
        self,
        base_url: str,
        timeout: float = 30.0,
        max_connections: int = 100,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.connector = aiohttp.TCPConnector(
            limit=max_connections,
            limit_per_host=20,
            ttl_dns_cache=300,
        )
        self._session: aiohttp.ClientSession | None = None

    async def __aenter__(self) -> "AsyncAPIClient":
        """Create session on context entry."""
        self._session = aiohttp.ClientSession(
            connector=self.connector,
            timeout=self.timeout,
        )
        return self

    async def __aexit__(self, *args: Any) -> None:
        """Close session on context exit."""
        if self._session:
            await self._session.close()
            self._session = None

    @retry(
        retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)),
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
    )
    async def get(self, path: str) -> APIResponse:
        """GET request with automatic retry."""
        if not self._session:
            raise RuntimeError("Client not initialized. Use async with.")

        url = f"{self.base_url}/{path.lstrip('/')}"

        async with self._session.get(url) as response:
            data = await response.json()
            return APIResponse(
                status=response.status,
                data=data,
                elapsed_ms=0.0,  # Would calculate from start time
            )

    async def get_many(
        self,
        paths: list[str],
        concurrency: int = 10,
    ) -> list[APIResponse | Exception]:
        """Fetch multiple paths with limited concurrency."""
        semaphore = asyncio.Semaphore(concurrency)

        async def fetch_one(path: str) -> APIResponse | Exception:
            async with semaphore:
                try:
                    return await self.get(path)
                except Exception as e:
                    return e

        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_one(p)) for p in paths]

        return [t.result() for t in tasks]


async def main() -> None:
    """Demonstrate async HTTP client usage."""
    async with AsyncAPIClient("https://api.example.com") as client:
        # Single request
        response = await client.get("/users/1")
        print(f"Status: {response.status}")

        # Multiple concurrent requests
        paths = [f"/users/{i}" for i in range(1, 11)]
        results = await client.get_many(paths, concurrency=5)
        successes = [r for r in results if isinstance(r, APIResponse)]
        print(f"Fetched {len(successes)} users")


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

Key points:

• Use aiohttp.TCPConnector for connection pooling
• Semaphore limits concurrent requests
• tenacity provides declarative retry logic

---

Example 4: Thread Pool for Blocking I/O

Use case: Run blocking operations without blocking the event loop.

#!/usr/bin/env python3
"""Mixing async with blocking I/O using thread pools."""

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path


def blocking_file_read(path: Path) -> str:
    """Blocking file operation (runs in thread pool)."""
    time.sleep(0.1)  # Simulate slow I/O
    return path.read_text() if path.exists() else ""


def cpu_intensive_task(data: str) -> int:
    """CPU-bound task (runs in thread pool)."""
    # Simulate CPU work
    total = sum(ord(c) for c in data)
    return total


async def process_files(paths: list[Path]) -> list[int]:
    """Process files using thread pool for blocking operations."""
    loop = asyncio.get_running_loop()
    results: list[int] = []

    # Create a dedicated thread pool
    with ThreadPoolExecutor(max_workers=4) as pool:
        # Read all files concurrently in threads
        read_tasks = [
            loop.run_in_executor(pool, blocking_file_read, path)
            for path in paths
        ]
        contents = await asyncio.gather(*read_tasks)

        # Process contents in threads
        process_tasks = [
            loop.run_in_executor(pool, cpu_intensive_task, content)
            for content in contents
        ]
        results = await asyncio.gather(*process_tasks)

    return results


async def simple_blocking_call() -> str:
    """Use asyncio.to_thread for simple blocking calls."""
    # Python 3.9+ convenience function
    result = await asyncio.to_thread(blocking_file_read, Path("test.txt"))
    return result


async def mixed_async_and_sync() -> None:
    """Combine async and sync operations."""
    # Async operation
    await asyncio.sleep(0.1)

    # Blocking operation in thread
    data = await asyncio.to_thread(time.sleep, 0.1)

    # More async operations
    async with asyncio.TaskGroup() as tg:
        tg.create_task(asyncio.sleep(0.1))
        tg.create_task(asyncio.to_thread(time.sleep, 0.1))


if __name__ == "__main__":
    # Process multiple files
    paths = [Path(f"file_{i}.txt") for i in range(5)]
    results = asyncio.run(process_files(paths))
    print(f"Processed {len(results)} files")

Key points:

• asyncio.to_thread() is the simplest way to run blocking code
• run_in_executor() allows custom thread/process pools
• Never call blocking code directly in async functions

---

Common Patterns

Pattern: Async Context Manager

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_resource():
    resource = await acquire_resource()
    try:
        yield resource
    finally:
        await release_resource(resource)

async def use_resource():
    async with managed_resource() as r:
        await r.do_work()

Pattern: Rate Limiting with Semaphore

async def rate_limited_fetch(urls: list[str], max_concurrent: int = 5):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch(url: str):
        async with semaphore:
            return await do_fetch(url)

    return await asyncio.gather(*[fetch(url) for url in urls])

Pattern: Cancellation Handling

async def cancellable_operation():
    try:
        while True:
            await asyncio.sleep(1)
            # Do work
    except asyncio.CancelledError:
        # Cleanup before propagating
        await cleanup()
        raise

---

Pitfalls to Avoid

Don't do this:

# Blocking the event loop
async def bad_example():
    time.sleep(1)  # Blocks entire event loop!
    data = requests.get(url)  # Also blocking!

Do this instead:

async def good_example():
    await asyncio.sleep(1)  # Non-blocking
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            data = await resp.json()

---

Don't do this:

# Fire and forget tasks (may be garbage collected)
async def bad_fire_and_forget():
    asyncio.create_task(some_background_work())
    # Task might be GC'd before completion!

Do this instead:

# Keep references to background tasks
background_tasks = set()

async def good_fire_and_forget():
    task = asyncio.create_task(some_background_work())
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)

---

See Also

• error-handling.md - Exception groups and async error handling
• context-managers.md - Async context managers
• fastapi-patterns.md - Async web frameworks