Python Engineering Excellence

This skill provides comprehensive Python engineering guidelines for modern Python development. Use this when writing or reviewing Python code for production systems, CLI tools, and AI agents.

Core Philosophy

The Zen of Python (Selected)

• Explicit is better than implicit - Make intentions clear

• Simple is better than complex - Favor straightforward solutions

• Readability counts - Code is read more than written

• Errors should never pass silently - Handle or propagate errors explicitly

• There should be one obvious way to do it - Follow established patterns

Design Principles

• Pass dependencies explicitly, avoid global state

• Favor composition over inheritance

• Keep functions small and focused

• Make side effects obvious

Project Structure

Standard Layout

project/ src/ mypackage/ init.py core.py cli.py agents/ init.py graph.py tests/ init.py test_core.py conftest.py pyproject.toml README.md

Guidelines:

• Use src/ layout to avoid import confusion

• One package per project under src/

• Tests mirror source structure

• All configuration in pyproject.toml

pyproject.toml (uv)

[project] name = "mypackage" version = "0.1.0" description = "My package description" requires-python = ">=3.11" dependencies = [ "typer>=0.9.0", "pydantic>=2.0", ]

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

[project.scripts] mycli = "mypackage.cli:app"

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

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

[tool.mypy] python_version = "3.11" warn_return_any = true warn_unused_ignores = true

[tool.pytest.ini_options] asyncio_mode = "auto" testpaths = ["tests"]

Naming Conventions

PEP 8 Rules

Good

module_name # modules: lowercase_with_underscores ClassName # classes: CamelCase function_name # functions: lowercase_with_underscores variable_name # variables: lowercase_with_underscores CONSTANT_NAME # constants: UPPERCASE_WITH_UNDERSCORES _private_var # private: leading underscore __mangled # name mangling: double underscore

Bad

ClassName() # Don't use for functions functionName # Don't use camelCase for functions VariableName # Don't use CamelCase for variables

Specific Conventions

Good - descriptive names

def calculate_total_price(items: list[Item]) -> Decimal: ...

user_count = len(users) is_valid = check_validity(data)

Good - short names for small scopes

for i, item in enumerate(items): process(item)

Bad - single letters for large scopes

def process(d): # What is d? ...

Boolean Naming

Good - question-like predicates

is_active = True has_permission = user.can_edit should_retry = attempt < max_retries

Bad

active = True # Unclear if boolean permission = True # Noun, not predicate

Code Organization

Imports

Use import statements for packages and modules, not for individual classes or functions (except from typing and collections.abc ).

Good - grouped and sorted (ruff handles this)

from future import annotations

import asyncio import os from pathlib import Path

import httpx from pydantic import BaseModel

from mypackage.core import process from mypackage.models import User

Good - import modules, not individual items

from sound.effects import echo echo.EchoFilter(...)

Good - typing symbols can be imported directly

from typing import Any, TypeVar from collections.abc import Sequence, Mapping

Good - use aliases when needed

import pandas as pd # Standard abbreviations OK from mypackage.submodule import very_long_module as vlm

Bad - importing individual classes (non-typing)

from sound.effects.echo import EchoFilter

Bad - ungrouped, relative imports in library code

from .core import process # Avoid relative imports import os, sys # Never multiple imports on one line

Module Structure

Good - clear all for public API

all = ["Client", "process_data", "DataError"]

class DataError(Exception): """Raised when data processing fails."""

class Client: """HTTP client for the service.""" ...

def process_data(data: bytes) -> dict: """Process raw data into structured format.""" ...

Private helpers below public API

def _validate(data: bytes) -> bool: ...

Class Organization

class Service: """Service for processing requests."""

def __init__(self, client: Client, config: Config) -> None:
    self._client = client
    self._config = config

# Public methods first
def process(self, request: Request) -> Response:
    data = self._fetch(request)
    return self._transform(data)

# Private methods after
def _fetch(self, request: Request) -> bytes:
    ...

def _transform(self, data: bytes) -> Response:
    ...

Error Handling

Exception Patterns

Good - custom exceptions with context

class ProcessingError(Exception): """Raised when processing fails."""

def __init__(self, message: str, item_id: str) -> None:
    super().__init__(message)
    self.item_id = item_id

Good - raise with context

def process(item: Item) -> Result: try: return transform(item) except ValueError as e: raise ProcessingError(f"failed to transform: {e}", item.id) from e

Good - use built-in exceptions appropriately

def set_age(age: int) -> None: if age < 0: raise ValueError("age must be non-negative")

Bad - bare except (catches KeyboardInterrupt, SystemExit)

try: process(item) except: # Never do this pass

Bad - catching Exception without re-raising

try: process(item) except Exception: pass # Silently swallowing errors

OK - catching Exception only if re-raising or at isolation point

try: process(item) except Exception: logger.exception("Processing failed") raise # Re-raise after logging

Assertions

Do not use assert for validation or preconditions—use explicit conditionals:

Bad - assert can be disabled with -O flag

def process(data: bytes) -> dict: assert data, "data required" # Skipped in optimized mode! return parse(data)

Good - explicit validation

def process(data: bytes) -> dict: if not data: raise ValueError("data required") return parse(data)

OK - assert in tests (pytest)

def test_process(): result = process(b"test") assert result["status"] == "ok"

Context Managers

Good - use context managers for cleanup

from contextlib import contextmanager

@contextmanager def managed_connection(dsn: str): conn = connect(dsn) try: yield conn finally: conn.close()

Usage

with managed_connection(dsn) as conn: conn.execute(query)

Good - async context manager

from contextlib import asynccontextmanager

@asynccontextmanager async def managed_session(): session = aiohttp.ClientSession() try: yield session finally: await session.close()

Guard Clauses

Good - early returns reduce nesting

def process_user(user: User | None) -> Result: if user is None: raise ValueError("user required")

if not user.is_active:
    return Result.inactive()

if not user.has_permission("process"):
    return Result.forbidden()

# Main logic at base indentation
return do_processing(user)

Bad - deeply nested

def process_user(user: User | None) -> Result: if user is not None: if user.is_active: if user.has_permission("process"): return do_processing(user) else: return Result.forbidden() else: return Result.inactive() else: raise ValueError("user required")

Async/Await

Basic Patterns

Good - async function

async def fetch_data(url: str) -> dict: async with httpx.AsyncClient() as client: response = await client.get(url) return response.json()

Good - gather for concurrent operations

async def fetch_all(urls: list[str]) -> list[dict]: async with httpx.AsyncClient() as client: tasks = [client.get(url) for url in urls] responses = await asyncio.gather(*tasks) return [r.json() for r in responses]

Bad - sequential when concurrent is possible

async def fetch_all_slow(urls: list[str]) -> list[dict]: results = [] for url in urls: data = await fetch_data(url) # Sequential! results.append(data) return results

Task Management

Good - structured concurrency with TaskGroup (Python 3.11+)

async def process_items(items: list[Item]) -> list[Result]: results = [] async with asyncio.TaskGroup() as tg: for item in items: task = tg.create_task(process_item(item)) results.append(task) return [t.result() for t in results]

Good - timeout handling

async def fetch_with_timeout(url: str, timeout: float = 30.0) -> dict: async with asyncio.timeout(timeout): return await fetch_data(url)

Async Context

Good - async generators

async def stream_results(query: str): async with get_connection() as conn: async for row in conn.execute(query): yield process_row(row)

Usage

async for result in stream_results(query): handle(result)

Type Hints

Gradual Typing Approach

Type hints encouraged but not enforced. Prioritize:

• Public API functions and methods

• Function signatures (args + return)

• Complex data structures

• Code that benefits from IDE support

Good - typed public API

def process_items(items: list[Item], *, strict: bool = False) -> list[Result]: """Process items and return results.""" ...

OK - internal helper without full typing

def _transform(data): # Complex internal logic ...

Common Patterns

from typing import TypeVar, Protocol, Callable from collections.abc import Iterator, Sequence

Generic types

T = TypeVar("T")

def first(items: Sequence[T]) -> T | None: return items[0] if items else None

Protocols for structural typing

class Processor(Protocol): def process(self, data: bytes) -> dict: ...

def run(processor: Processor, data: bytes) -> dict: return processor.process(data)

Callable types

Handler = Callable[[Request], Response]

def with_logging(handler: Handler) -> Handler: def wrapper(request: Request) -> Response: log(request) return handler(request) return wrapper

Union and Optional (use | syntax in 3.10+)

def find_user(user_id: str) -> User | None: ...

TypedDict for structured dicts

from typing import TypedDict

class UserData(TypedDict): name: str email: str age: int | None

Type Narrowing

Good - type narrowing with isinstance

def process(value: str | int) -> str: if isinstance(value, str): return value.upper() # Type narrowed to str return str(value)

Good - assert for narrowing (use sparingly)

def process_user(user: User | None) -> str: assert user is not None, "user required" return user.name # Type narrowed to User

Formatting

Line Length and Indentation

• Maximum 80 characters per line (URLs and long imports excepted)

• Use 4 spaces for indentation; never tabs

• Use implicit line continuation inside parentheses, brackets, braces

Good - implicit continuation with aligned elements

result = some_function( argument_one, argument_two, argument_three, )

Good - hanging indent

result = some_function( argument_one, argument_two, argument_three, )

Bad - backslash continuation

result = some_long_function_name()
+ another_function()

Good - parentheses for continuation

result = ( some_long_function_name() + another_function() )

Whitespace

Good

spam(ham[1], {eggs: 2}) x = 1 dict["key"] = list[index] def func(default: str = "value") -> None: ...

Bad - spaces inside brackets

spam( ham[ 1 ], { eggs: 2 } )

Bad - space before bracket

spam (ham[1]) dict ["key"]

Good - break at highest syntactic level

if ( condition_one and condition_two and condition_three ): do_something()

Bad - break in middle of expression

if (condition_one and condition_two): do_something()

Blank Lines

• Two blank lines between top-level definitions (functions, classes)

• One blank line between method definitions

• No blank line after def line

Comprehensions

Use comprehensions for simple transformations. Avoid complex comprehensions.

Good - simple comprehension

squares = [x * x for x in range(10)] evens = {x for x in numbers if x % 2 == 0} mapping = {k: v for k, v in pairs}

Good - generator for large data

total = sum(x * x for x in range(1000000))

Bad - multiple for clauses (hard to read)

result = [ (x, y, z) for x in range(5) for y in range(5) for z in range(5) if x != y ]

Good - use nested loops instead

result = [] for x in range(5): for y in range(5): for z in range(5): if x != y: result.append((x, y, z))

Bad - complex conditions in comprehension

result = [transform(x) for x in data if validate(x) and x.enabled and x.value > 0]

Good - extract to function or use loop

def should_include(x): return validate(x) and x.enabled and x.value > 0

result = [transform(x) for x in data if should_include(x)]

Strings

Formatting

Use f-strings for interpolation. For logging, use % format with pattern strings.

Good - f-strings for general use

message = f"Processing {item.name} (id={item.id})"

Good - logging with % patterns (deferred formatting)

logger.info("Processing %s (id=%s)", item.name, item.id)

Bad - f-strings in logging (always formatted even if not logged)

logger.debug(f"Data: {expensive_repr(data)}")

Good - join for concatenation in loops

parts = [] for item in items: parts.append(str(item)) result = ", ".join(parts)

Or simply:

result = ", ".join(str(item) for item in items)

Bad - += concatenation in loop

result = "" for item in items: result += str(item) + ", " # Creates many intermediate strings

Multiline Strings

Good - textwrap.dedent for indented multiline

import textwrap

long_string = textwrap.dedent("""
First line Second line Third line """)

Good - implicit concatenation

message = ( "This is a very long message that needs " "to be split across multiple lines for " "readability purposes." )

Configuration

Pydantic Settings

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings): model_config = SettingsConfigDict( env_prefix="MYAPP_", env_file=".env", )

database_url: str
api_key: str
debug: bool = False
max_workers: int = 4

Usage

settings = Settings() # Loads from env vars / .env

CLI Arguments with Typer

import typer from typing import Annotated

app = typer.Typer()

@app.command() def main( input_file: Annotated[Path, typer.Argument(help="Input file path")], output: Annotated[Path, typer.Option("--output", "-o")] = Path("out.json"), verbose: Annotated[bool, typer.Option("--verbose", "-v")] = False, ) -> None: """Process input file and write results.""" if verbose: print(f"Processing {input_file}")

result = process(input_file)
output.write_text(json.dumps(result))

if name == "main": app()

Testing

Pytest Basics

tests/test_core.py

import pytest from mypackage.core import process, ProcessingError

def test_process_valid_input(): result = process(b"valid data") assert result["status"] == "ok"

def test_process_empty_input(): with pytest.raises(ProcessingError) as exc_info: process(b"") assert "empty" in str(exc_info.value)

Fixtures

tests/conftest.py

import pytest from mypackage import Client, Config

@pytest.fixture def config() -> Config: return Config(api_url="http://test", timeout=1.0)

@pytest.fixture def client(config: Config) -> Client: return Client(config)

Fixture with cleanup

@pytest.fixture def temp_db(tmp_path): db_path = tmp_path / "test.db" db = create_database(db_path) yield db db.close()

Async fixtures

@pytest.fixture async def async_client(): async with httpx.AsyncClient() as client: yield client

Parametrized Tests

@pytest.mark.parametrize( "input_data,expected", [ (b"hello", {"word": "hello", "length": 5}), (b"world", {"word": "world", "length": 5}), (b"", None), ], ids=["hello", "world", "empty"], ) def test_parse(input_data: bytes, expected: dict | None): result = parse(input_data) assert result == expected

Mocking

from unittest.mock import Mock, AsyncMock, patch

def test_with_mock(): mock_client = Mock() mock_client.fetch.return_value = {"data": "test"}

service = Service(client=mock_client)
result = service.process()

mock_client.fetch.assert_called_once()
assert result == {"data": "test"}

Async mock

async def test_async_service(): mock_client = AsyncMock() mock_client.fetch.return_value = {"data": "test"}

service = AsyncService(client=mock_client)
result = await service.process()

assert result == {"data": "test"}

Patching

@patch("mypackage.core.external_api") def test_with_patch(mock_api): mock_api.return_value = "mocked" result = function_using_api() assert result == "mocked"

CLI Development

Typer Application Structure

src/mypackage/cli.py

import typer from rich.console import Console

app = typer.Typer(help="My CLI application") console = Console()

@app.command() def init( name: str, force: bool = typer.Option(False, "--force", "-f", help="Overwrite existing"), ) -> None: """Initialize a new project.""" if Path(name).exists() and not force: console.print(f"[red]Error:[/red] {name} already exists") raise typer.Exit(1)

create_project(name)
console.print(f"[green]Created[/green] {name}")

@app.command() def run( config: Path = typer.Option(Path("config.yaml"), "--config", "-c"), ) -> None: """Run the application.""" settings = load_config(config) execute(settings)

if name == "main": app()

Subcommands

Main app with subcommands

app = typer.Typer() db_app = typer.Typer(help="Database commands") app.add_typer(db_app, name="db")

@db_app.command("migrate") def db_migrate(): """Run database migrations.""" ...

@db_app.command("seed") def db_seed(): """Seed database with test data.""" ...

Usage: mycli db migrate

AI Agents / LangGraph

State Management

from typing import Annotated, TypedDict from langgraph.graph import StateGraph from langgraph.graph.message import add_messages

class AgentState(TypedDict): messages: Annotated[list, add_messages] context: dict next_step: str | None

def create_agent() -> StateGraph: graph = StateGraph(AgentState)

graph.add_node("process", process_node)
graph.add_node("decide", decision_node)
graph.add_node("execute", execute_node)

graph.add_edge("process", "decide")
graph.add_conditional_edges(
    "decide",
    route_decision,
    {"execute": "execute", "end": "__end__"},
)
graph.add_edge("execute", "process")

graph.set_entry_point("process")
return graph.compile()

Tool Definitions

from langchain_core.tools import tool

@tool def search_database(query: str) -> list[dict]: """Search the database for matching records.

Args:
    query: Search query string

Returns:
    List of matching records
"""
return db.search(query)

@tool async def fetch_url(url: str) -> str: """Fetch content from a URL.

Args:
    url: URL to fetch

Returns:
    Page content as text
"""
async with httpx.AsyncClient() as client:
    response = await client.get(url)
    return response.text

Node Functions

from langchain_core.messages import HumanMessage, AIMessage

async def process_node(state: AgentState) -> AgentState: """Process incoming messages and update context.""" last_message = state["messages"][-1]

if isinstance(last_message, HumanMessage):
    # Process user input
    context = await analyze_message(last_message.content)
    return {"context": context}

return {}

def decision_node(state: AgentState) -> AgentState: """Decide next action based on context.""" context = state["context"]

if context.get("needs_execution"):
    return {"next_step": "execute"}

return {"next_step": "end"}

def route_decision(state: AgentState) -> str: """Route based on decision.""" return state.get("next_step", "end")

Dependency Injection

Constructor Injection

Good - dependencies passed explicitly

class UserService: def init(self, db: Database, cache: Cache, logger: Logger) -> None: self._db = db self._cache = cache self._logger = logger

def get_user(self, user_id: str) -> User | None:
    if cached := self._cache.get(user_id):
        return cached

    user = self._db.find_user(user_id)
    if user:
        self._cache.set(user_id, user)
    return user

Bad - hidden dependencies

class UserService: def get_user(self, user_id: str) -> User | None: return global_db.find_user(user_id) # Hidden dependency

Factory Functions

def create_service(config: Config) -> Service: """Create service with all dependencies configured.""" db = Database(config.database_url) cache = RedisCache(config.redis_url) logger = setup_logger(config.log_level)

return Service(db=db, cache=cache, logger=logger)

In tests

def create_test_service() -> Service: return Service( db=InMemoryDatabase(), cache=DictCache(), logger=NullLogger(), )

Main Entry Point

Always guard module-level code to prevent execution on import:

Good - guarded entry point

def main() -> None: """Application entry point.""" config = load_config() result = process(config) print(result)

if name == "main": main()

With CLI framework (typer)

import typer

app = typer.Typer()

@app.command() def main(config: Path = typer.Option(...)) -> None: """Process with configuration.""" ...

if name == "main": app()

Anti-patterns to Avoid

Mutable Default Arguments

Bad - mutable default shared across calls

def append_item(item, items=[]): items.append(item) return items

Good - use None and create new list

def append_item(item, items=None): if items is None: items = [] items.append(item) return items

Bare Except

Bad - catches everything including KeyboardInterrupt

try: process() except: pass

Good - catch specific exceptions

try: process() except ProcessingError as e: logger.error(f"Processing failed: {e}") raise

Star Imports

Bad - pollutes namespace, unclear origins

from module import *

Good - explicit imports

from module import SpecificClass, specific_function

Overusing Inheritance

Bad - deep inheritance hierarchy

class Animal: ... class Mammal(Animal): ... class Dog(Mammal): ... class GermanShepherd(Dog): ...

Good - composition and protocols

class Animal(Protocol): def speak(self) -> str: ...

class Dog: def init(self, breed: str) -> None: self.breed = breed

def speak(self) -> str:
    return "woof"

God Classes

Bad - class doing too much

class Application: def connect_database(self): ... def send_email(self): ... def process_payment(self): ... def generate_report(self): ... def authenticate_user(self): ...

Good - single responsibility

class DatabaseConnection: ... class EmailService: ... class PaymentProcessor: ... class ReportGenerator: ... class AuthService: ...

Avoid Power Features

Avoid metaclasses, dynamic attribute access via getattr , bytecode manipulation, and reflection tricks. Use simpler alternatives.

Bad - metaclass for simple use case

class SingletonMeta(type): _instances = {} def call(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super().call(*args, **kwargs) return cls._instances[cls]

Good - module-level instance or factory function

_instance = None

def get_instance() -> Service: global _instance if _instance is None: _instance = Service() return _instance

Avoid staticmethod

Use module-level functions instead of @staticmethod :

Bad - staticmethod

class StringUtils: @staticmethod def clean(text: str) -> str: return text.strip().lower()

Good - module-level function

def clean_text(text: str) -> str: return text.strip().lower()

Boolean Evaluation Pitfalls

Good - explicit None check

if value is not None: process(value)

Bad - falsy check when 0 or "" are valid

if value: # Fails for value=0 or value="" process(value)

Good - explicit comparison for sequences

if len(items) == 0: return default

OK - implicit boolean for sequences (when falsy means empty)

if not items: return default

Documentation

Docstrings

def process_data( data: bytes, *, encoding: str = "utf-8", strict: bool = False, ) -> dict: """Process raw bytes into structured data.

Args:
    data: Raw bytes to process
    encoding: Text encoding to use
    strict: If True, raise on invalid data

Returns:
    Parsed data as dictionary

Raises:
    ProcessingError: If data is invalid and strict=True

Example:
    >>> result = process_data(b'{"key": "value"}')
    >>> result["key"]
    'value'
"""
...

Class Docstrings

class DataProcessor: """Process and transform data from various sources.

This class handles reading data from files or URLs,
validating the format, and transforming it into
the required output structure.

Attributes:
    config: Processor configuration
    stats: Processing statistics

Example:
    >>> processor = DataProcessor(Config())
    >>> result = processor.process(data)
"""

def __init__(self, config: Config) -> None:
    """Initialize processor with configuration.

    Args:
        config: Processor configuration object
    """
    self.config = config
    self.stats = Stats()

Tooling Workflow

uv Commands

Create new project

uv init myproject cd myproject

Add dependencies

uv add httpx pydantic typer

Add dev dependencies

uv add --dev pytest pytest-asyncio mypy ruff

Sync dependencies

uv sync

Run commands in venv

uv run python script.py uv run pytest uv run mypy src/

Build package

uv build

ruff Commands

Check for issues

uv run ruff check .

Fix auto-fixable issues

uv run ruff check --fix .

Format code

uv run ruff format .

Check formatting without changes

uv run ruff format --check .

mypy Commands

Type check

uv run mypy src/

Strict mode (if desired)

uv run mypy --strict src/

Generate stub files

uv run stubgen -p mypackage

pytest Commands

Run all tests

uv run pytest

Verbose with output

uv run pytest -v -s

Run specific test

uv run pytest tests/test_core.py::test_process

Run with coverage

uv run pytest --cov=mypackage --cov-report=html

Run async tests (with pytest-asyncio)

uv run pytest # asyncio_mode = "auto" in pyproject.toml

Pre-commit Workflow

1. Format

uv run ruff format .

2. Lint and fix

uv run ruff check --fix .

3. Type check

uv run mypy src/

4. Run tests

uv run pytest

5. All checks pass, ready to commit

Additional Resources

• Google Python Style Guide

• PEP 8 - Style Guide

• PEP 484 - Type Hints

• uv Documentation

• ruff Documentation

• pytest Documentation

• LangGraph Documentation

• Typer Documentation

Remember: These guidelines support writing clear, maintainable Python code. Adapt them to your specific context, but always favor readability, explicitness, and simplicity.