Python – Best Practices

This page covers the programming practices that produce production-grade, idiomatic Python at Cygnus Dynamics. These go beyond PEP 8 formatting to cover how to write code that is correct, readable, testable, safe, and performant.

Docstrings

Write docstrings for all public modules, classes, methods, and functions. Use the Google docstring style consistently across all Cygnus Dynamics Python projects — it renders cleanly in Sphinx and is readable inline.

Module Docstrings

"""Order service — manages the full order lifecycle.

This module provides the OrderService class which coordinates
order creation, payment processing, and fulfilment notifications.
It should be the single entry point for all order-related
business operations.
"""

Function and Method Docstrings

def calculate_discounted_total(
    items: list[OrderItem],
    discount_rate: float,
    currency: str = "USD",
) -> float:
    """Calculate the total cost of an order after applying a discount.

    The discount is applied to the subtotal before tax. Tax calculation
    is handled separately by the TaxService based on the billing address.

    Args:
        items: The order items to price. Must not be empty.
        discount_rate: Fractional discount to apply (0.0 to 1.0).
            A value of 0.20 applies a 20% discount.
        currency: ISO 4217 currency code. Defaults to "USD".

    Returns:
        The discounted subtotal rounded to 2 decimal places.

    Raises:
        EmptyOrderError: If ``items`` is empty.
        ValueError: If ``discount_rate`` is outside the range [0.0, 1.0].

    Example:
        >>> items = [OrderItem(price=50.0, quantity=2)]
        >>> calculate_discounted_total(items, discount_rate=0.10)
        90.0
    """
    if not items:
        raise EmptyOrderError("Cannot calculate total for empty order")
    if not 0.0 <= discount_rate <= 1.0:
        raise ValueError(f"discount_rate must be between 0 and 1, got {discount_rate}")

    subtotal = sum(item.price * item.quantity for item in items)
    return round(subtotal * (1 - discount_rate), 2)

Class Docstrings

class OrderService:
    """Manages the full lifecycle of customer orders.

    This service coordinates with PaymentService and InventoryService
    to ensure all order operations are transactionally consistent.
    Use this class as the single entry point for all order operations.

    Attributes:
        repository: Persistent store for order records.
        payment_service: Handles payment authorisation and capture.

    Example:
        >>> service = OrderService(repository, payment_service)
        >>> order = service.create_order(request)
    """

    def __init__(
        self,
        repository: OrderRepository,
        payment_service: PaymentService,
    ) -> None:
        self._repository = repository
        self._payment_service = payment_service

One-liner docstrings — for simple functions, a single line is sufficient. Keep the closing """ on the same line:

def is_active(user: User) -> bool:
    """Return True if the user account is currently active."""
    return user.status == UserStatus.ACTIVE

Error Handling

Use Specific Exceptions

Always raise and catch the most specific exception type available. Catching Exception masks unexpected failures.

# ✅ Specific exception — caller knows exactly what went wrong
def get_order(self, order_id: int) -> Order:
    order = self._repository.find_by_id(order_id)
    if order is None:
        raise OrderNotFoundError(order_id)
    return order

# ✅ Specific catch — handles the expected case, re-raises unexpected ones
try:
    order = order_service.get_order(order_id)
except OrderNotFoundError:
    return {"error": "Order not found"}, 404

# ❌ Bare except — catches KeyboardInterrupt, SystemExit, and everything else
try:
    order = order_service.get_order(order_id)
except:
    return {"error": "failed"}, 500

# ❌ Too broad — catches TypeError, AttributeError, and every other bug
try:
    order = order_service.get_order(order_id)
except Exception:
    return {"error": "failed"}, 500

Context Managers for Resources

Always use with statements for resources that must be closed — files, database connections, locks, HTTP sessions. This ensures cleanup even if an exception is raised.

# ✅ with statement — file is always closed
def read_report(path: Path) -> str:
    with open(path, encoding="utf-8") as f:
        return f.read()

# ✅ Multiple resources in one with (Python 3.10+ parenthesised form)
def merge_reports(input_path: Path, output_path: Path) -> None:
    with (
        open(input_path, encoding="utf-8") as source,
        open(output_path, "w", encoding="utf-8") as dest,
    ):
        dest.write(source.read())

# ❌ Manual close — won't run if exception is raised before it
def read_report(path: Path) -> str:
    f = open(path)
    content = f.read()
    f.close()      # never reached if read() raises
    return content

Never Use Exceptions for Flow Control

Exceptions signal unexpected, abnormal conditions. Do not use them to control normal program flow.

# ❌ Exception used as a boolean check — slow and misleading
def user_exists(user_id: int) -> bool:
    try:
        user_repository.get(user_id)
        return True
    except UserNotFoundError:
        return False

# ✅ Use a purpose-built method
def user_exists(user_id: int) -> bool:
    return user_repository.exists(user_id)

Preserve Exception Context

When raising a new exception inside an except block, always chain the original using raise ... from:

# ✅ Correct — original traceback preserved
try:
    result = db.execute(query, params)
except psycopg2.OperationalError as exc:
    raise DatabaseConnectionError("Failed to connect to orders DB") from exc

# ❌ Incorrect — original traceback lost
try:
    result = db.execute(query, params)
except psycopg2.OperationalError:
    raise DatabaseConnectionError("Failed to connect to orders DB")

Async / Await

Use async/await for all I/O-bound operations in services built on async frameworks (FastAPI, aiohttp, Starlette). Mixing sync and async code incorrectly is a common source of bugs and performance problems.

Core Rules

• Declare a function async def only when it contains at least one await expression or calls another async function.
• Never call a blocking I/O operation (file I/O, requests.get, time.sleep) directly inside an async def — it blocks the entire event loop.
• Use asyncio.gather() to run independent async operations concurrently rather than awaiting them sequentially.
• Use asyncio.create_task() when you want to fire and not await a background operation.

Correct Async Patterns

import asyncio
import httpx
from sqlalchemy.ext.asyncio import AsyncSession


# ✅ Async service method with async DB session
async def get_order(self, order_id: int) -> Order:
    result = await self._session.execute(
        select(Order).where(Order.id == order_id)
    )
    order = result.scalar_one_or_none()
    if order is None:
        raise OrderNotFoundError(order_id)
    return order


# ✅ Concurrent I/O with asyncio.gather — runs both calls simultaneously
async def get_order_with_user(order_id: int) -> tuple[Order, User]:
    order, user = await asyncio.gather(
        order_service.get_order(order_id),
        user_service.get_user(order.user_id),
    )
    return order, user


# ✅ Async context manager for HTTP client
async def fetch_shipping_rates(postcode: str) -> list[ShippingRate]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            f"{SHIPPING_API_URL}/rates",
            params={"postcode": postcode},
        )
        response.raise_for_status()
        return [ShippingRate(**r) for r in response.json()]


# ✅ Background task — fire and forget
async def process_order(order: Order) -> None:
    await order_repository.save(order)
    asyncio.create_task(analytics_service.track_order_created(order))  # non-blocking

What Not to Do

import time
import requests


# ❌ Blocking sleep inside async function — freezes the event loop
async def retry_with_backoff(func, retries: int = 3) -> None:
    for attempt in range(retries):
        try:
            await func()
            return
        except TransientError:
            time.sleep(2 ** attempt)  # BLOCKS the event loop — use asyncio.sleep

# ✅ Non-blocking sleep
async def retry_with_backoff(func, retries: int = 3) -> None:
    for attempt in range(retries):
        try:
            await func()
            return
        except TransientError:
            await asyncio.sleep(2 ** attempt)  # yields control back to the event loop


# ❌ Sync HTTP call inside async function — blocks the event loop
async def fetch_product(product_id: int) -> dict:
    response = requests.get(f"{PRODUCT_API}/products/{product_id}")  # sync — blocks
    return response.json()

# ✅ Use an async HTTP client
async def fetch_product(product_id: int) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{PRODUCT_API}/products/{product_id}")
        response.raise_for_status()
        return response.json()


# ❌ Sequential awaits when operations are independent
async def enrich_order(order: Order) -> EnrichedOrder:
    user = await user_service.get_user(order.user_id)     # waits for this
    product = await product_service.get(order.product_id) # then waits for this
    return EnrichedOrder(order, user, product)

# ✅ Concurrent — both fetched at the same time
async def enrich_order(order: Order) -> EnrichedOrder:
    user, product = await asyncio.gather(
        user_service.get_user(order.user_id),
        product_service.get(order.product_id),
    )
    return EnrichedOrder(order, user, product)

Running Sync Code from Async Context

When you must call a blocking/sync function from async code (e.g. a legacy library), run it in a thread pool executor to avoid blocking the event loop:

import asyncio
from functools import partial


# ✅ Run blocking code in a thread pool — does not block event loop
async def generate_pdf_report(order_id: int) -> bytes:
    loop = asyncio.get_running_loop()
    pdf_bytes = await loop.run_in_executor(
        None,  # use the default ThreadPoolExecutor
        partial(pdf_generator.render, order_id=order_id),
    )
    return pdf_bytes

Pythonic Patterns

Truthiness

Use Python's built-in truthiness — do not compare to True, False, None, 0, or [] explicitly unless you need to distinguish between falsy values.

# ✅ Pythonic truthiness checks
if items:                     # True if list is non-empty
    process(items)

if not user.is_active:        # more readable than == False
    raise InactiveUserError()

if config is None:            # use `is` to check for None
    config = default_config()

# ❌ Explicit comparison to booleans or zero
if len(items) > 0:            # just use `if items:`
    process(items)

if user.is_active == True:    # just use `if user.is_active:`
    pass

if config == None:            # use `is None`
    pass

Enumerate and Zip

# ✅ enumerate — when you need both index and value
for index, item in enumerate(order.items, start=1):
    print(f"{index}. {item.name} — {item.price}")

# ✅ zip — iterate two sequences in parallel
for item, quantity in zip(products, quantities):
    cart.add(item, quantity)

# ❌ Manual index tracking
for i in range(len(order.items)):
    item = order.items[i]       # verbose — use enumerate instead

List Comprehensions vs Loops

Use list, dict, and set comprehensions for simple, single-expression transformations. For complex logic, use a for loop with an explicit append or a named helper function.

# ✅ Simple filter and map
active_emails = [user.email for user in users if user.is_active]
order_totals  = [order.total for order in orders]

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

# ✅ Set comprehension — unique values
unique_currencies = {order.currency for order in orders}

# ✅ For complex logic — use a named function, not a nested comprehension
def extract_valid_items(order: Order) -> list[OrderItem]:
    valid_items = []
    for item in order.items:
        if item.quantity > 0 and item.is_in_stock():
            valid_items.append(item)
    return valid_items

# ❌ Nested comprehension — unreadable
result = [
    transform(item)
    for order in orders
    for item in order.items
    if item.is_valid()
    if order.is_active
]

Generator Expressions for Large Data

When you do not need a materialised list — for example, when passing to sum(), any(), all(), or iterating once — use a generator expression. It avoids building the full list in memory.

# ✅ Generator — does not build a full list
total = sum(item.price * item.quantity for item in order.items)
has_out_of_stock = any(not item.is_in_stock() for item in order.items)
all_confirmed    = all(o.status == "CONFIRMED" for o in orders)

# ❌ List comprehension when a generator would do — wastes memory
total = sum([item.price * item.quantity for item in order.items])

Avoid Mutable Default Arguments

Never use a mutable object (list, dict, set) as a default argument — it is shared across all calls.

# ❌ Mutable default — the same list is shared across all calls
def add_item(item: str, cart: list = []) -> list:
    cart.append(item)
    return cart

add_item("apple")   # ["apple"]
add_item("banana")  # ["apple", "banana"] — unexpected!

# ✅ Use None and create inside the function
def add_item(item: str, cart: list | None = None) -> list:
    if cart is None:
        cart = []
    cart.append(item)
    return cart

Functions and Classes

Guard Clauses Over Deep Nesting

Return or raise early to keep the happy path at the top level.

# ❌ Deep nesting
def process_order(order: Order, user: User) -> None:
    if order:
        if order.items:
            if user.is_active:
                if user.has_payment_method:
                    charge_and_fulfil(order, user)

# ✅ Guard clauses — flat and readable
def process_order(order: Order, user: User) -> None:
    if not order:
        raise ValueError("Order is required")
    if not order.items:
        raise EmptyOrderError(order.id)
    if not user.is_active:
        raise InactiveUserError(user.id)
    if not user.has_payment_method:
        raise NoPaymentMethodError(user.id)

    charge_and_fulfil(order, user)

Properties Over Direct Attribute Access

Use @property to provide controlled access to internal state while keeping the external API clean:

class Order:

    def __init__(self, items: list[OrderItem]) -> None:
        self._items = items
        self._status = OrderStatus.PENDING

    @property
    def total(self) -> float:
        """The order total, calculated from current items."""
        return sum(item.price * item.quantity for item in self._items)

    @property
    def status(self) -> OrderStatus:
        return self._status

    def confirm(self) -> None:
        if self._status != OrderStatus.PENDING:
            raise InvalidStateTransitionError(self._status, OrderStatus.CONFIRMED)
        self._status = OrderStatus.CONFIRMED

Comments

Block Comments

Block comments should be indented to the same level as the code they describe. Write complete sentences starting with a capital letter.

# ✅ Block comment — complete sentence, explains WHY
def charge_order(order: Order) -> PaymentResult:
    # We use idempotency keys to prevent double-charging if the request
    # times out and is retried. The key is scoped to the order ID.
    idempotency_key = f"charge_{order.id}"
    return payment_gateway.charge(order.total, idempotency_key)

Inline Comments

Use inline comments sparingly — only when the code alone cannot communicate intent.

# ✅ Inline comment explains a non-obvious value
SESSION_TIMEOUT_SEC = 1800  # 30 minutes — per security policy SP-12

# ❌ Narrates the obvious — remove it
x = x + 1  # Increment x by 1

TODO and FIXME Markers

Always include a ticket reference. A TODO without a ticket is invisible to planning.

# ✅ Traceable and actionable
# TODO(PROJ-1234): Replace with async notification when event bus is ready
email_service.send_sync(user.email, message)

# FIXME(PROJ-5678): Refund calculation does not account for partial returns
refund_total = order.total

# ❌ No ticket — untrackable, will never be fixed
# TODO: fix this later

Programming Practices

Comparisons

# ✅ Use `is` and `is not` for singletons
if result is None:
    return default
if flag is not False:
    proceed()

# ✅ Use isinstance() for type checks — not type()
if isinstance(value, str):
    process_string(value)

# ❌ type() check — fails for subclasses
if type(value) == str:
    pass

String Formatting

Use f-strings for all string interpolation in Python 3.6+.

# ✅ f-strings — readable, fast, modern
message = f"Hello {name}, your order #{order_id} has been confirmed."
log_entry = f"[{datetime.now().isoformat()}] Order {order_id} processed"

# ❌ Old-style formatting
message = "Hello %s, your order #%d has been confirmed." % (name, order_id)
message = "Hello {}, your order #{} has been confirmed.".format(name, order_id)

__all__ for Public APIs

Define __all__ in every module intended to be imported by others. This explicitly declares the public API and prevents from module import * from leaking internal names.

# order_service.py
__all__ = ["OrderService", "CreateOrderRequest", "OrderNotFoundError"]


class OrderService:
    ...


class CreateOrderRequest:
    ...


class OrderNotFoundError(Exception):
    ...


def _internal_helper() -> None:    # not in __all__ — not part of public API
    ...

Testing

Test File Naming

order_service.py         →  test_order_service.py
user_repository.py       →  test_user_repository.py
payment_processor.py     →  test_payment_processor.py

Test Structure — Arrange / Act / Assert

import pytest
from unittest.mock import MagicMock

from cygnus.orders.service import OrderService
from cygnus.orders.exceptions import OrderNotFoundError, EmptyOrderError


class TestOrderService:

    def setup_method(self) -> None:
        self.repository = MagicMock()
        self.payment_service = MagicMock()
        self.service = OrderService(self.repository, self.payment_service)

    def test_get_order_returns_order_when_found(self) -> None:
        # Arrange
        expected = Order(id=1, user_id=42, total=99.99)
        self.repository.find_by_id.return_value = expected

        # Act
        result = self.service.get_order(order_id=1)

        # Assert
        assert result == expected
        self.repository.find_by_id.assert_called_once_with(1)

    def test_get_order_raises_when_not_found(self) -> None:
        # Arrange
        self.repository.find_by_id.return_value = None

        # Act & Assert
        with pytest.raises(OrderNotFoundError) as exc_info:
            self.service.get_order(order_id=999)

        assert exc_info.value.order_id == 999

    def test_create_order_raises_for_empty_cart(self) -> None:
        # Arrange
        request = CreateOrderRequest(user_id=1, items=[])

        # Act & Assert
        with pytest.raises(EmptyOrderError):
            self.service.create_order(request)

Async Tests

Use pytest-asyncio for testing async code. Annotate async test functions with @pytest.mark.asyncio.

import pytest
import pytest_asyncio
from unittest.mock import AsyncMock


class TestAsyncOrderService:

    @pytest_asyncio.fixture
    async def service(self) -> AsyncOrderService:
        repository = AsyncMock()
        return AsyncOrderService(repository)

    @pytest.mark.asyncio
    async def test_get_order_returns_order_when_found(
        self, service: AsyncOrderService
    ) -> None:
        # Arrange
        expected = Order(id=1, user_id=42)
        service._repository.find_by_id.return_value = expected

        # Act
        result = await service.get_order(order_id=1)

        # Assert
        assert result == expected

    @pytest.mark.asyncio
    async def test_get_order_raises_when_not_found(
        self, service: AsyncOrderService
    ) -> None:
        service._repository.find_by_id.return_value = None

        with pytest.raises(OrderNotFoundError):
            await service.get_order(order_id=999)

Coverage expectations:

Scope	Target
Service layer business logic	90%+
Utility functions	100%
Repository / DB layer	Integration tests
API endpoints	Integration tests covering all status codes

Project Configuration

Every Python project at Cygnus Dynamics must commit these configuration files to the repository root. Do not rely on default settings — explicit config ensures all developers and CI run identical checks.

pyproject.toml

[tool.black]
line-length = 99
target-version = ["py311"]

[tool.isort]
profile = "black"
line_length = 99
known_first_party = ["cygnus"]

[tool.mypy]
python_version = "3.11"
strict = true
ignore_missing_imports = true
disallow_untyped_defs = true
disallow_any_generics = true
warn_return_any = true
warn_unused_ignores = true

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
addopts = "--strict-markers --tb=short -q"

[tool.coverage.run]
source = ["cygnus"]
omit = ["*/tests/*", "*/migrations/*"]

[tool.coverage.report]
fail_under = 80
show_missing = true
skip_covered = false

[tool.ruff]
line-length = 99
target-version = "py311"
select = ["E", "F", "W", "I", "N", "UP", "B", "C4", "SIM"]
ignore = ["E501"]

.pre-commit-config.yaml

repos:
  - repo: https://github.com/psf/black
    rev: 24.3.0
    hooks:
      - id: black
        language_version: python3.11

  - repo: https://github.com/PyCQA/isort
    rev: 5.13.2
    hooks:
      - id: isort

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

  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.9.0
    hooks:
      - id: mypy
        additional_dependencies: [types-requests, types-PyYAML]

  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.6.0
    hooks:
      - id: trailing-whitespace
      - id: end-of-file-fixer
      - id: check-yaml
      - id: check-merge-conflict
      - id: debug-statements     # catches leftover pdb/print calls

Makefile — common commands

.PHONY: format lint typecheck test

format:
    black .
    isort .

lint:
    ruff check .
    flake8 .

typecheck:
    mypy cygnus/

test:
    pytest --cov=cygnus --cov-report=term-missing

check: format lint typecheck test

Tooling Quick Reference

Tool	Purpose	Command
black	Auto-format all files	black .
isort	Sort and group imports	isort .
ruff	Fast linter — replaces flake8 + many plugins	ruff check .
flake8	PEP 8 lint check (legacy projects)	flake8 .
pylint	Deeper static analysis	pylint cygnus/
mypy	Static type checking	mypy cygnus/
pytest	Run tests with coverage	pytest --cov=cygnus
pre-commit	Run all hooks manually	pre-commit run --all-files

Install pre-commit hooks once per clone

After cloning any Python repository, run pre-commit install to activate the hooks. This ensures formatting and linting run automatically on every commit — no manual step required before pushing.