Python Expert

Specialized expertise in Python programming, providing deep guidance on type annotations and static analysis, exception handling idioms, async/await concurrency, data modeling with dataclasses and Pydantic, testing with pytest, performance optimization, and modern packaging with pyproject.toml.

When to Use This Skill

Use this skill for:

Adding type annotations and configuring mypy strict mode
Designing custom exception hierarchies and error chains
Implementing async/await patterns with asyncio
Modeling data with dataclasses and Pydantic
Writing pytest test suites with fixtures and parametrize
Profiling and optimizing Python performance
Setting up Python project structure and packaging

Trigger phrases: "python", "asyncio", "pydantic", "pytest", "mypy", "type hints", "dataclass", "python packaging", "python performance"

What This Skill Does

Provides Python expertise including:

Type System: Annotations, generics, protocols, mypy strict
Error Handling: Custom exceptions, error chains, context managers
Async/Await: asyncio event loop, task groups, async generators
Data Modeling: dataclasses, Pydantic BaseModel, validation
Testing: pytest fixtures, parametrize, monkeypatch, coverage
Performance: Profiling, generators, slots, caching, comprehensions
Packaging: pyproject.toml, src layout, virtual environments

Instructions

Step 1: Master Type Annotations and mypy-Strict Patterns

Enable Postponed Evaluation and Modern Syntax:

from __future__ import annotations

from typing import TypeVar, Generic, Protocol

# X | Y union syntax (works at runtime with __future__ annotations)
def fetch_user(user_id: int) -> User | None:
    """Return user or None if not found."""
    row = db.query("SELECT * FROM users WHERE id = %s", (user_id,))
    if row is None:
        return None
    return User.from_row(row)

# Avoid Optional[X] and Union[X, Y] in new code
# Bad:  Optional[str]
# Good: str | None
def parse_header(raw: str) -> str | None:
    parts = raw.split(":", maxsplit=1)
    return parts[1].strip() if len(parts) == 2 else None

TypeVar and Generic Containers:

from typing import TypeVar, Generic

T = TypeVar("T")

class Stack(Generic[T]):
    """Type-safe generic stack."""

    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        if not self._items:
            raise IndexError("pop from empty stack")
        return self._items.pop()

    def peek(self) -> T:
        if not self._items:
            raise IndexError("peek at empty stack")
        return self._items[-1]

    def __len__(self) -> int:
        return len(self._items)

# Usage: mypy enforces type consistency
stack: Stack[int] = Stack()
stack.push(42)
stack.push("oops")  # mypy error: Argument 1 has incompatible type "str"

Protocol for Structural Subtyping:

from typing import Protocol, runtime_checkable

@runtime_checkable
class Serializable(Protocol):
    def to_dict(self) -> dict[str, object]: ...

class User:
    def __init__(self, name: str, age: int) -> None:
        self.name = name
        self.age = age

    def to_dict(self) -> dict[str, object]:
        return {"name": self.name, "age": self.age}

# User satisfies Serializable without inheriting from it
def serialize(obj: Serializable) -> str:
    import json
    return json.dumps(obj.to_dict())

# Runtime check also works
assert isinstance(User("Alice", 30), Serializable)

mypy Strict Configuration (pyproject.toml):

[tool.mypy]
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_any_generics = true
check_untyped_defs = true

Step 2: Handle Exceptions Idiomatically

Custom Exception Hierarchy:

class AppError(Exception):
    """Base exception for the application."""

    def __init__(self, message: str, code: str = "UNKNOWN") -> None:
        super().__init__(message)
        self.code = code

class NotFoundError(AppError):
    """Raised when a resource is not found."""

    def __init__(self, resource: str, identifier: object) -> None:
        super().__init__(
            f"{resource} with id {identifier!r} not found",
            code="NOT_FOUND",
        )
        self.resource = resource
        self.identifier = identifier

class ValidationError(AppError):
    """Raised when input validation fails."""

    def __init__(self, field: str, reason: str) -> None:
        super().__init__(
            f"Validation failed for {field!r}: {reason}",
            code="VALIDATION_ERROR",
        )
        self.field = field
        self.reason = reason

Error Chaining with raise-from:

import logging

logger = logging.getLogger(__name__)

def load_config(path: str) -> dict[str, object]:
    try:
        with open(path) as f:
            import json
            return json.load(f)
    except FileNotFoundError as exc:
        raise AppError(f"Config file missing: {path}") from exc
    except json.JSONDecodeError as exc:
        raise AppError(f"Invalid JSON in {path}: line {exc.lineno}") from exc

# The original exception is preserved in __cause__
# Tracebacks show: "The above exception was the direct cause of..."

Context Managers for Resource Cleanup:

from contextlib import contextmanager
from typing import Iterator

@contextmanager
def temporary_directory(prefix: str = "tmp") -> Iterator[str]:
    """Create a temp directory and clean it up on exit."""
    import tempfile
    import shutil

    path = tempfile.mkdtemp(prefix=prefix)
    try:
        yield path
    finally:
        shutil.rmtree(path, ignore_errors=True)

# Usage
with temporary_directory("build_") as build_dir:
    compile_assets(build_dir)
    # Directory is cleaned up even if an exception occurs

# Class-based context manager for database transactions
class Transaction:
    def __init__(self, conn: Connection) -> None:
        self._conn = conn

    def __enter__(self) -> Transaction:
        self._conn.begin()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: object,
    ) -> bool:
        if exc_type is not None:
            self._conn.rollback()
            logger.error("Transaction rolled back: %s", exc_val)
            return False  # Re-raise the exception
        self._conn.commit()
        return False

Step 3: Implement Async/Await Patterns

Basic asyncio Patterns:

import asyncio

async def fetch_url(session: aiohttp.ClientSession, url: str) -> str:
    """Fetch a single URL with timeout."""
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
        resp.raise_for_status()
        return await resp.text()

async def fetch_all(urls: list[str]) -> list[str]:
    """Fetch multiple URLs concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=False)

# Entry point
async def main() -> None:
    urls = [
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    ]
    results = await fetch_all(urls)
    for url, body in zip(urls, results):
        print(f"{url}: {len(body)} bytes")

asyncio.run(main())

Task Groups (Python 3.11+):

async def process_batch(items: list[Item]) -> list[Result]:
    """Process items concurrently with structured cancellation."""
    results: list[Result] = []

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(process_item(item)) for item in items]

    # All tasks completed successfully if we reach here.
    # If any task raises, the group cancels all remaining tasks
    # and raises an ExceptionGroup.
    results = [task.result() for task in tasks]
    return results

Async Context Managers and Generators:

from contextlib import asynccontextmanager
from typing import AsyncIterator

@asynccontextmanager
async def managed_connection(dsn: str) -> AsyncIterator[Connection]:
    """Acquire a database connection and release it on exit."""
    conn = await asyncpg.connect(dsn)
    try:
        yield conn
    finally:
        await conn.close()

# Async generator for streaming results
async def stream_rows(query: str) -> AsyncIterator[dict[str, object]]:
    """Yield rows one at a time without loading all into memory."""
    async with managed_connection(DATABASE_URL) as conn:
        async with conn.transaction():
            async for record in conn.cursor(query):
                yield dict(record)

# Consuming an async generator
async def export_csv(query: str, path: str) -> int:
    count = 0
    with open(path, "w") as f:
        async for row in stream_rows(query):
            f.write(",".join(str(v) for v in row.values()) + "\n")
            count += 1
    return count

Semaphore for Rate Limiting:

async def fetch_with_limit(
    urls: list[str],
    max_concurrent: int = 10,
) -> list[str]:
    """Fetch URLs with a concurrency limit."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_fetch(session: aiohttp.ClientSession, url: str) -> str:
        async with semaphore:
            return await fetch_url(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [limited_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

Step 4: Model Data with Dataclasses and Pydantic

Dataclass Fundamentals:

from dataclasses import dataclass, field
from datetime import datetime

@dataclass(frozen=True)
class Money:
    """Immutable value object for monetary amounts."""
    amount: int  # Store as cents to avoid float issues
    currency: str = "USD"

    def __post_init__(self) -> None:
        if self.amount < 0:
            raise ValueError("Amount cannot be negative")

    def __add__(self, other: Money) -> Money:
        if self.currency != other.currency:
            raise ValueError(f"Cannot add {self.currency} and {other.currency}")
        return Money(self.amount + other.amount, self.currency)

@dataclass
class Order:
    """Mutable entity with computed defaults."""
    customer_id: int
    items: list[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.utcnow)
    total: Money = field(default_factory=lambda: Money(0))

    def add_item(self, name: str, price: Money) -> None:
        self.items.append(name)
        self.total = self.total + price

Dataclass with Slots (Python 3.10+):

@dataclass(slots=True)
class Point:
    """Memory-efficient dataclass using __slots__."""
    x: float
    y: float
    z: float = 0.0

    def distance_to(self, other: Point) -> float:
        return (
            (self.x - other.x) ** 2
            + (self.y - other.y) ** 2
            + (self.z - other.z) ** 2
        ) ** 0.5

Pydantic BaseModel for Validation and Serialization:

from pydantic import BaseModel, Field, field_validator, model_validator

class CreateUserRequest(BaseModel):
    """Validated API request model."""
    username: str = Field(min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$")
    email: str = Field(max_length=255)
    age: int = Field(ge=13, le=150)
    tags: list[str] = Field(default_factory=list, max_length=10)

    @field_validator("email")
    @classmethod
    def validate_email(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("Invalid email address")
        return v.lower()

    @model_validator(mode="after")
    def check_consistency(self) -> CreateUserRequest:
        if self.age < 18 and "admin" in self.tags:
            raise ValueError("Users under 18 cannot be admins")
        return self

# Parse and validate from dict (e.g., request.json())
user = CreateUserRequest.model_validate({"username": "alice", "email": "ALICE@example.com", "age": 25})
print(user.email)  # alice@example.com (lowercased by validator)

# Serialize to dict or JSON
user.model_dump()
user.model_dump_json()

Pydantic Settings for Configuration:

from pydantic_settings import BaseSettings

class AppSettings(BaseSettings):
    """Load settings from environment variables."""
    database_url: str
    redis_url: str = "redis://localhost:6379"
    debug: bool = False
    max_connections: int = 10

    model_config = {"env_prefix": "APP_"}

# Reads APP_DATABASE_URL, APP_REDIS_URL, etc. from environment
settings = AppSettings()

Step 5: Write Thorough Tests with pytest

Basic Test Structure and Parametrize:

import pytest
from myapp.calculator import add, divide

def test_add_positive_numbers() -> None:
    assert add(2, 3) == 5

def test_divide_by_zero_raises() -> None:
    with pytest.raises(ZeroDivisionError, match="division by zero"):
        divide(10, 0)

@pytest.mark.parametrize(
    ("a", "b", "expected"),
    [
        (2, 3, 5),
        (-1, -1, -2),
        (0, 0, 0),
        (-1, 1, 0),
        (100, 200, 300),
    ],
    ids=["positive", "negative", "zeros", "mixed", "large"],
)
def test_add_parametrized(a: int, b: int, expected: int) -> None:
    assert add(a, b) == expected

Fixtures and conftest.py:

# conftest.py
import pytest
from myapp.database import Database

@pytest.fixture
def db(tmp_path: object) -> Database:
    """Create a fresh test database for each test."""
    db = Database(f"sqlite:///{tmp_path}/test.db")
    db.create_tables()
    yield db
    db.close()

@pytest.fixture
def sample_user(db: Database) -> User:
    """Insert and return a sample user."""
    return db.create_user(name="Alice", email="alice@example.com")

# test_users.py
def test_user_creation(db: Database) -> None:
    user = db.create_user(name="Bob", email="bob@example.com")
    assert user.name == "Bob"
    assert user.id is not None

def test_user_lookup(db: Database, sample_user: User) -> None:
    found = db.get_user(sample_user.id)
    assert found is not None
    assert found.email == "alice@example.com"

Monkeypatch for Isolation:

def test_config_reads_env(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("APP_DATABASE_URL", "sqlite:///test.db")
    monkeypatch.setenv("APP_DEBUG", "true")

    settings = AppSettings()
    assert settings.database_url == "sqlite:///test.db"
    assert settings.debug is True

def test_fetch_retries_on_failure(monkeypatch: pytest.MonkeyPatch) -> None:
    call_count = 0

    def mock_get(url: str, timeout: int = 30) -> object:
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise ConnectionError("network error")
        return MockResponse(status_code=200, body="ok")

    monkeypatch.setattr("myapp.client.requests.get", mock_get)

    result = fetch_with_retry("https://api.example.com/data")
    assert result == "ok"
    assert call_count == 3

Async Test Support:

import pytest

@pytest.mark.asyncio
async def test_async_fetch() -> None:
    async with aiohttp.ClientSession() as session:
        result = await fetch_url(session, "https://httpbin.org/get")
        assert len(result) > 0

# pytest.ini or pyproject.toml:
# [tool.pytest.ini_options]
# asyncio_mode = "auto"

Coverage Configuration (pyproject.toml):

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--strict-markers -q"

[tool.coverage.run]
source = ["src"]
branch = true

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

Step 6: Optimize Performance

Profiling Before Optimizing:

import cProfile
import pstats

def profile_function(func, *args, **kwargs):
    """Profile a function and print the top 20 calls by cumulative time."""
    profiler = cProfile.Profile()
    profiler.enable()
    result = func(*args, **kwargs)
    profiler.disable()

    stats = pstats.Stats(profiler)
    stats.sort_stats("cumulative")
    stats.print_stats(20)
    return result

# Or use the decorator approach
# python -m cProfile -s cumulative myapp/main.py

Generators for Memory Efficiency:

# Bad: loads everything into memory
def read_all_lines(path: str) -> list[str]:
    with open(path) as f:
        return f.readlines()  # Entire file in memory

# Good: yields one line at a time
def read_lines(path: str) -> Iterator[str]:
    with open(path) as f:
        for line in f:
            yield line.strip()

# Generator expression (lazy evaluation)
total = sum(len(line) for line in read_lines("large_file.txt"))

# itertools for composable lazy pipelines
import itertools

def process_large_dataset(path: str) -> Iterator[dict[str, str]]:
    lines = read_lines(path)
    non_empty = (line for line in lines if line)
    batches = itertools.batched(non_empty, 1000)  # Python 3.12+
    for batch in batches:
        yield from process_batch(batch)

Slots for Memory-Efficient Classes:

class HeavyObject:
    """Default: uses __dict__ for attribute storage."""
    def __init__(self, x: int, y: int, z: int) -> None:
        self.x = x
        self.y = y
        self.z = z

class LightObject:
    """Slots: fixed attribute set, ~40% less memory per instance."""
    __slots__ = ("x", "y", "z")

    def __init__(self, x: int, y: int, z: int) -> None:
        self.x = x
        self.y = y
        self.z = z

# Memory comparison (approximate):
# HeavyObject: ~152 bytes per instance
# LightObject: ~88 bytes per instance

Caching with lru_cache and cache:

from functools import lru_cache, cache

@lru_cache(maxsize=256)
def fibonacci(n: int) -> int:
    """Memoized Fibonacci: O(n) instead of O(2^n)."""
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# Unbounded cache (Python 3.9+)
@cache
def expensive_lookup(key: str) -> dict[str, object]:
    return database.query(key)

# Cache with TTL using third-party library
from cachetools import TTLCache

api_cache: TTLCache[str, dict[str, object]] = TTLCache(maxsize=1000, ttl=300)

Comprehensions Over Loops:

# List comprehension (faster than append loop)
squares = [x * x for x in range(1000)]

# Dict comprehension
index = {user.id: user for user in users}

# Set comprehension for deduplication
unique_domains = {email.split("@")[1] for email in emails}

# Conditional comprehension
active_users = [u for u in users if u.is_active and u.last_login is not None]

# Avoid nested comprehensions beyond 2 levels; use a function instead
# Bad:
matrix = [[row[i] for row in data] for i in range(cols)]
# Better:
def transpose(data: list[list[int]]) -> list[list[int]]:
    return [list(col) for col in zip(*data)]

Step 7: Structure Projects and Packaging

Modern pyproject.toml Configuration:

[project]
name = "myapp"
version = "1.2.0"
description = "A production application"
requires-python = ">=3.11"
license = {text = "MIT"}
dependencies = [
    "pydantic>=2.0,<3.0",
    "httpx>=0.25",
    "asyncpg>=0.29",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-cov>=5.0",
    "pytest-asyncio>=0.23",
    "mypy>=1.8",
    "ruff>=0.3",
]

[project.scripts]
myapp = "myapp.cli:main"

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

[tool.ruff]
line-length = 88
target-version = "py311"

[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "SIM", "TCH"]

src Layout (Recommended):

myapp/
  pyproject.toml
  src/
    myapp/
      __init__.py
      cli.py
      config.py
      models/
        __init__.py
        user.py
      services/
        __init__.py
        auth.py
      utils/
        __init__.py
        logging.py
  tests/
    unit/
      test_models.py
      test_services.py
    integration/
      test_database.py
    conftest.py

Virtual Environment Management with uv:

# Create virtual environment
uv venv

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

# Add a new dependency
uv pip install httpx

# Sync from lock file (reproducible installs)
uv pip sync requirements.lock

# Generate lock file
uv pip compile pyproject.toml -o requirements.lock

Package Entry Points and CLI:

# src/myapp/cli.py
from __future__ import annotations

import argparse
import sys

def main(argv: list[str] | None = None) -> int:
    """Application entry point."""
    parser = argparse.ArgumentParser(description="My Application")
    parser.add_argument("--config", default="config.toml", help="Config file path")
    parser.add_argument("--verbose", action="store_true", help="Enable verbose output")
    subparsers = parser.add_subparsers(dest="command", required=True)

    serve_parser = subparsers.add_parser("serve", help="Start the server")
    serve_parser.add_argument("--port", type=int, default=8080)

    migrate_parser = subparsers.add_parser("migrate", help="Run database migrations")
    migrate_parser.add_argument("--dry-run", action="store_true")

    args = parser.parse_args(argv)

    if args.command == "serve":
        return run_server(args.port, args.config)
    elif args.command == "migrate":
        return run_migrations(args.config, dry_run=args.dry_run)

    return 0

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

Best Practices

Type everything - Run mypy strict in CI; zero type errors is the target
Raise specific exceptions - Never raise bare Exception; define a project hierarchy
Use context managers - For any resource that needs cleanup (files, connections, locks)
Prefer composition - Inherit only from abstract base classes or Protocol
Immutable by default - Use frozen dataclasses and tuples unless mutation is required
Profile first - Measure before optimizing; use cProfile or py-spy
Test at boundaries - Mock external services, not internal functions
Pin dependencies - Exact versions in production; ranges only in library pyproject.toml

Common Patterns

Pattern 1: Repository Pattern with Protocol

from __future__ import annotations

from typing import Protocol

class UserRepository(Protocol):
    def get(self, user_id: int) -> User | None: ...
    def save(self, user: User) -> User: ...
    def delete(self, user_id: int) -> bool: ...

class PostgresUserRepository:
    def __init__(self, conn: asyncpg.Connection) -> None:
        self._conn = conn

    async def get(self, user_id: int) -> User | None:
        row = await self._conn.fetchrow(
            "SELECT * FROM users WHERE id = $1", user_id
        )
        return User(**dict(row)) if row else None

    async def save(self, user: User) -> User:
        row = await self._conn.fetchrow(
            "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING *",
            user.name,
            user.email,
        )
        return User(**dict(row))

    async def delete(self, user_id: int) -> bool:
        result = await self._conn.execute(
            "DELETE FROM users WHERE id = $1", user_id
        )
        return result == "DELETE 1"

class InMemoryUserRepository:
    """Test double that satisfies the same Protocol."""

    def __init__(self) -> None:
        self._store: dict[int, User] = {}
        self._next_id = 1

    def get(self, user_id: int) -> User | None:
        return self._store.get(user_id)

    def save(self, user: User) -> User:
        user.id = self._next_id
        self._store[self._next_id] = user
        self._next_id += 1
        return user

    def delete(self, user_id: int) -> bool:
        return self._store.pop(user_id, None) is not None

Pattern 2: Retry with Exponential Backoff

from __future__ import annotations

import asyncio
import logging
from typing import TypeVar

T = TypeVar("T")
logger = logging.getLogger(__name__)

async def retry(
    func,
    *args,
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    retryable: tuple[type[Exception], ...] = (ConnectionError, TimeoutError),
) -> T:
    """Retry an async callable with exponential backoff."""
    last_exc: Exception | None = None

    for attempt in range(1, max_attempts + 1):
        try:
            return await func(*args)
        except retryable as exc:
            last_exc = exc
            if attempt == max_attempts:
                break
            delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
            logger.warning(
                "Attempt %d/%d failed (%s), retrying in %.1fs",
                attempt,
                max_attempts,
                exc,
                delay,
            )
            await asyncio.sleep(delay)

    raise last_exc  # type: ignore[misc]

# Usage
result = await retry(fetch_url, session, url, max_attempts=5)

Quality Checklist

All functions have type annotations (parameters and return)
mypy strict passes with zero errors
Custom exceptions inherit from a project base class
Context managers used for resource cleanup
Async code uses TaskGroup or gather for concurrency
Pydantic models validate all external input
pytest coverage at or above 80%
No mutable default arguments in function signatures
ruff check and ruff format pass
pyproject.toml defines all project metadata

Related Skills

performance-testing - Python profiling and benchmarks
cicd-architect - Python CI/CD pipelines
code-quality - Python code standards
api-designer - FastAPI and async web services

Version: 1.0.0 Last Updated: March 2026 Based on: awesome-codex-subagents python-pro patterns

Iterative Refinement Strategy

This skill is optimized for an iterative approach:

Execute: Perform the core steps defined above.
Review: Critically analyze the output (coverage, quality, completeness).
Refine: If targets are not met, repeat the specific implementation steps with improved context.
Loop: Continue until the definition of done is satisfied.

python-expert

SKILL.mdskills/common/python-expert/SKILL.md