Skip to content

Python Coding Standards ​

These coding standards define how we write, organize, and maintain our Python projects. By following them, we ensure our codebase remains consistent, predictable, and easy to evolve.

Philosophy ​

Our primary goals are:

  • Consistency: Every file should look and behave predictably across the project.
  • Clarity: Code should express its intent clearly, without unnecessary complexity.
  • Type Safety: Strong typing prevents subtle bugs and simplifies refactoring.
  • Maintainability: Future developers should understand the system without relying on tribal knowledge.
  • AI-Friendly: Clear code standards help Generative AI tools better understand and generate code that aligns with our project conventions.
  • Reusability: Factor out common logic (logger setup, health probes, LLM usage) into reusable pip packages and use this code in the packages.

Tooling ​

We use the following tools to maintain high quality and performance:

  • Astral ty: For type checking.
  • Astral uv: For Python package and project management.
  • Astral ruff: For linting and code formatting.
  • HTTPX: For synchronous and asynchronous HTTP requests.
  • Returns: For making functions return something meaningful, typed, and safe.
  • Dependency Injector: For dependency injection.
  • FastAPI: For API creation.

General Principles ​

  • Do: Optimize for Python 3.13.
  • Do: Follow PEP 8 style guide.
  • Do: Write docstrings for modules, classes, and functions.
  • Do: Write small functions that do one thing.
  • Do: Write unit tests with as little mocking as possible.
  • Do: Use the Receive an Object, Return an Object (RORO) pattern where appropriate.
  • Do not: Have side effects in functions unless explicitly intended.
  • Do not: Write unnecessary comments; code should be self-documenting.

Docstrings ​

We use Google-style docstrings for all modules, classes, and functions.

  • Do: Write a one-line summary that fits on one line.
  • Do: Add a blank line after the summary if there are more sections.
  • Do: Document all arguments, return values, and raised exceptions.
  • Do: Use type hints in signatures; don't repeat types in docstrings.
python
# ✅ Right - Google-style docstring
def translate_text(
    text: str,
    source_lang: str,
    target_lang: str,
    *,
    preserve_formatting: bool = True,
) -> TranslationResult:
    """Translate text from source language to target language.

    Uses the configured LLM to perform translation while optionally
    preserving the original formatting.

    Args:
        text: The text to translate.
        source_lang: ISO 639-1 language code of the source language.
        target_lang: ISO 639-1 language code of the target language.
        preserve_formatting: Whether to maintain original text formatting.

    Returns:
        A TranslationResult containing the translated text and metadata.

    Raises:
        ValueError: If source_lang or target_lang is not a valid language code.
        TranslationError: If the translation service fails.

    Example:
        >>> result = translate_text("Hello", "en", "de")
        >>> print(result.text)
        "Hallo"
    """
python
# ✅ Right - Simple function with one-liner docstring
def get_user_by_id(user_id: str) -> User | None:
    """Retrieve a user by their unique identifier."""
    return user_repository.find_by_id(user_id)
python
# ✅ Right - Class docstring
class TranslationService:
    """Service for translating text between languages.

    This service wraps the LLM-based translation logic and provides
    caching and rate limiting capabilities.

    Attributes:
        model: The LLM model identifier used for translations.
        cache_enabled: Whether translation caching is active.
    """

    def __init__(self, config: AppConfig) -> None:
        """Initialize the translation service.

        Args:
            config: Application configuration containing LLM settings.
        """
        self.model = config.llm_model
        self.cache_enabled = True

Naming Conventions ​

  • Do: Use snake_case for variables and functions.
  • Do: Use PascalCase for classes.
  • Do: Use UPPER_CASE for constants.
  • Do: Use lower_case for modules.
python
# ✅ Right
class ShoppingCart:
    def calculate_total(self, items: list[Item]) -> float:
        return sum(item.price for item in items)

MAX_RETRIES = 3

# ❌ Wrong
class shoppingCart:
    def calcTotal(self, Items):
        pass

Code Style & Layout ​

  • Do: Use 4 spaces for indentation.
  • Do: Limit line length to 120 characters (configured in Ruff).
  • Do: Use f-strings for string formatting.
  • Do: Use is or is not for comparisons with None.
python
# ✅ Right
name = "Alice"
greeting = f"Hello, {name}!"

if user is None:
    return

# ❌ Wrong
greeting = "Hello, " + name + "!"

if user == None:
    return

Modern Python & Best Practices ​

  • Do: Use list comprehensions for simple transformations.
  • Do: Use pathlib over os.path for file system operations.
  • Do: Use enumerate and zip for cleaner loops.
  • Do: Use Type Hints everywhere, including for variables and return types.
python
# ✅ Right
squares = [x**2 for x in range(10)]
path = Path("data/file.txt")

for i, item in enumerate(items):
    print(f"{i}: {item}")

# ❌ Wrong
squares = []
for x in range(10):
    squares.append(x**2)
path = os.path.join("data", "file.txt")

Functions vs Classes ​

  • Do: Prefer functions over classes for stateless operations.
  • Do: Use dataclasses over regular classes for data containers.
  • Do: Use classes when you need to maintain state or implement interfaces.
  • Do not: Create classes with only __init__ and one method—use a function instead.
python
# ✅ Right - Use a function for stateless operations
def calculate_discount(price: float, discount_percent: float) -> float:
    """Calculate the discounted price."""
    return price * (1 - discount_percent / 100)


# ❌ Wrong - Unnecessary class for stateless operation
class DiscountCalculator:
    def calculate(self, price: float, discount_percent: float) -> float:
        return price * (1 - discount_percent / 100)
python
# ✅ Right - Use dataclass for data containers
from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class TranslationRequest:
    """Request payload for translation."""

    text: str
    source_lang: str
    target_lang: str
    preserve_formatting: bool = True


# ❌ Wrong - Regular class for simple data
class TranslationRequest:
    def __init__(
        self,
        text: str,
        source_lang: str,
        target_lang: str,
        preserve_formatting: bool = True,
    ) -> None:
        self.text = text
        self.source_lang = source_lang
        self.target_lang = target_lang
        self.preserve_formatting = preserve_formatting

Dataclass best practices:

  • Use frozen=True for immutable data (prevents accidental mutation).
  • Use slots=True for memory efficiency.
  • Use kw_only=True for classes with many fields to enforce named arguments.

Abstract Classes & Interfaces ​

Use abc.ABC and typing.Protocol to define interfaces and abstract base classes.

  • Do: Use Protocol for structural subtyping (duck typing with type safety).
  • Do: Use ABC when you need shared implementation or enforced inheritance.
  • Do: Place abstract classes in a dedicated interfaces/ or protocols/ module, or alongside related code.
python
# ✅ Right - Protocol for structural subtyping (preferred for interfaces)
from typing import Protocol


class Translatable(Protocol):
    """Interface for objects that can be translated."""

    def translate(self, target_lang: str) -> str:
        """Translate content to the target language."""
        ...


# Any class with a matching `translate` method satisfies this protocol
class Document:
    def translate(self, target_lang: str) -> str:
        return f"Translated to {target_lang}"


def process_translatable(item: Translatable) -> str:
    """Works with any object that has a translate method."""
    return item.translate("de")
python
# ✅ Right - ABC for shared implementation
from abc import ABC, abstractmethod


class AbstractAppConfig(ABC):
    """Base configuration class with shared functionality."""

    @classmethod
    @abstractmethod
    def from_env(cls) -> "AbstractAppConfig":
        """Load configuration from environment variables."""
        ...

    @abstractmethod
    def __str__(self) -> str:
        """Return a string representation (with secrets masked)."""
        ...

    def validate(self) -> bool:
        """Shared validation logic for all configs."""
        return True

When to use which:

Use CaseChoice
Define a contract/interfaceProtocol
Shared implementation neededABC
Duck typing with type safetyProtocol
Enforce inheritance hierarchyABC

Error Handling & Exceptions ​

  • Do: Handle errors and edge cases at the beginning of functions (guard clauses).
  • Do: Use early returns to avoid nested if/else structures.
  • Do: Catch specific exceptions (e.g., ValueError) rather than bare except.
  • Do: Use exception chaining (raise ... from e) to preserve the original traceback.
  • Do: Use context managers (with statement) for resource management.
  • Do not: Use bare except: clauses; this masks unexpected errors (like KeyboardInterrupt).
  • Do not: Suppress exceptions silently with pass unless absolutely necessary and documented.
  • Do not: Log an exception and then re-raise it (double logging).
python
# ✅ Right
def process_file(path: Path):
    try:
        with open(path, "r") as f:  # Context manager
            return json.load(f)
    except json.JSONDecodeError as e:
        logger.error("Invalid JSON format")
        raise DataError("Corrupt file") from e  # Exception chaining

# ❌ Wrong
def process_file_bad(path):
    try:
        f = open(path, "r")
        return json.load(f)
    except:  # Bare except
        logger.error("Something went wrong") # Double logging if re-raised
        raise

Functional Error Handling (Returns) ​

We use the Returns library to make our functions return meaningful, typed, and safe values. This helps avoid side effects and makes error handling explicit.

Result Container ​

Use Result (Success or Failure) for pure functions that can fail.

  • Do: Return Success(value) for successful operations.
  • Do: Return Failure(exception) for failed operations.
  • Do: Use @safe decorator to automatically wrap functions that might raise exceptions.
python
from returns.result import Result, Success, Failure, safe

@safe
def divide(a: int, b: int) -> float:
    return a / b

# Usage
result = divide(10, 2) # Success(5.0)
result = divide(10, 0) # Failure(ZeroDivisionError)

if isinstance(result, Success):
    print(f"Result: {result.unwrap()}")
else:
    print(f"Error: {result.failure()}")

IO and IOResult ​

Use IO for functions with side effects and IOResult for impure functions that can fail.

  • Do: Use IOResult for operations like HTTP requests or filesystem access.
  • Do: Use @impure_safe to wrap impure functions.
python
from returns.io import IOResult, impure_safe
import requests

@impure_safe
def fetch_user_data(user_id: int) -> dict:
    response = requests.get(f"https://api.example.com/users/{user_id}")
    response.raise_for_status()
    return response.json()

Future and FutureResult ​

Use Future and FutureResult for asynchronous operations.

  • Do: Use FutureResult for async operations that can fail (e.g., async HTTP calls).
  • Do: Use @future_safe to wrap async functions.
python
from returns.future import future_safe, FutureResult
import httpx

@future_safe
async def fetch_user_async(user_id: int) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        response.raise_for_status()
        return response.json()

Pipelines and Composition ​

Use flow, bind, and map to compose functions without manual error checking.

python
from returns.pipeline import flow
from returns.pointfree import bind

def process_user(user_id: int) -> FutureResult:
    return flow(
        user_id,
        fetch_user_async,
        bind(validate_user),
        bind(save_user_locally)
    )

Imports ​

  • Do: Group imports: Standard Library first, then Third Party, then Local Application.
  • Do not: Use wildcard imports (from module import *).
python
# ✅ Right
import sys
from pathlib import Path

import requests

from myapp.utils import helper

# ❌ Wrong
from myapp.utils import *

Async/Await Patterns ​

Since we use FastAPI, understanding async patterns is essential for building performant APIs.

When to Use Async ​

  • Do: Use async def for I/O-bound operations (HTTP calls, database queries, file I/O).
  • Do: Use regular def for CPU-bound operations (they run in a thread pool in FastAPI).
  • Do not: Mix blocking calls inside async functions without proper handling.
python
# ✅ Right - Async for I/O-bound operations
async def fetch_translation(text: str, target_lang: str) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.translation.com/translate",
            json={"text": text, "target": target_lang},
        )
        return response.json()["translation"]


# ✅ Right - Regular def for CPU-bound (FastAPI handles threading)
def compute_similarity(text_a: str, text_b: str) -> float:
    # CPU-intensive operation
    return calculate_levenshtein_distance(text_a, text_b)

Running Concurrent Tasks ​

Use asyncio.gather or asyncio.TaskGroup for concurrent operations:

python
import asyncio


# ✅ Right - Concurrent execution with gather
async def translate_batch(texts: list[str], target_lang: str) -> list[str]:
    tasks = [fetch_translation(text, target_lang) for text in texts]
    return await asyncio.gather(*tasks)


# ✅ Right - TaskGroup for structured concurrency (Python 3.11+)
async def process_documents(doc_ids: list[str]) -> list[Document]:
    results: list[Document] = []
    async with asyncio.TaskGroup() as tg:
        for doc_id in doc_ids:
            tg.create_task(fetch_and_process(doc_id, results))
    return results

Async Context Managers ​

python
from contextlib import asynccontextmanager


@asynccontextmanager
async def get_db_session():
    """Async context manager for database sessions."""
    session = await create_session()
    try:
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()


# Usage
async def save_translation(translation: Translation) -> None:
    async with get_db_session() as session:
        session.add(translation)

Avoiding Common Pitfalls ​

python
# ❌ Wrong - Blocking call in async function
async def bad_fetch():
    response = requests.get("https://api.example.com")  # Blocks the event loop!
    return response.json()


# ✅ Right - Use async HTTP client
async def good_fetch():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com")
        return response.json()


# ❌ Wrong - Sequential when could be concurrent
async def slow_batch():
    result1 = await fetch_data(1)
    result2 = await fetch_data(2)  # Waits for result1 unnecessarily
    return [result1, result2]


# ✅ Right - Concurrent execution
async def fast_batch():
    return await asyncio.gather(fetch_data(1), fetch_data(2))

HTTP Clients (HTTPX) ​

We use HTTPX as our primary HTTP client. It provides a modern, fully featured API for both synchronous and asynchronous requests.

Sync vs Async Usage ​

Always prefer AsyncClient (async) over Client (sync) whenever possible, as it provides better scalability and allows for concurrent execution without blocking.

Choose the appropriate client based on your execution context:

  • Use httpx.Client for synchronous code, scripts, or when performing sequential requests in a blocking environment.
  • Use httpx.AsyncClient for asynchronous code, especially within FastAPI routers or when you need to perform multiple requests concurrently.
python
# ✅ Right - Synchronous usage
import httpx

def fetch_data_sync(url: str) -> dict:
    """Fetch data from a URL synchronously."""
    with httpx.Client(timeout=10.0) as client:
        response = client.get(url)
        response.raise_for_status()
        return response.json()


# ✅ Right - Asynchronous usage
import httpx

async def fetch_data_async(url: str) -> dict:
    """Fetch data from a URL asynchronously."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

Best Practices ​

  • Do: Reuse client instances. Creating a new client for every request is expensive as it requires setting up a new connection pool and SSL context. In services, inject the client or manage it as a long-lived object.
  • Do: Always use context managers. Use with or async with to ensure connections are properly closed and returned to the pool.
  • Do: Set explicit timeouts. Never use default timeouts (which can be infinite in some clients). Always specify a reasonable timeout for your use case.
  • Do: Use response.raise_for_status(). This ensures that HTTP errors (4xx and 5xx) are raised as exceptions, preventing your code from proceeding with invalid data.
  • Do: Enable HTTP/2 for performance. If the server supports it, HTTP/2 can significantly improve performance through multiplexing.
  • Do: Configure limits for high-concurrency. Use httpx.Limits to control the maximum number of open connections and keep-alive connections.
python
# ✅ Right - Efficient concurrent requests with shared client
import asyncio
import httpx

async def fetch_all(urls: list[str]) -> list[dict]:
    """Fetch data from multiple URLs concurrently using a single client."""
    # Configure limits and timeouts globally for the client
    limits = httpx.Limits(max_connections=100, max_keepalive_connections=20)
    timeout = httpx.Timeout(10.0, connect=5.0)

    async with httpx.AsyncClient(timeout=timeout, limits=limits, http2=True) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)

        for response in responses:
            response.raise_for_status()

        return [response.json() for response in responses]
  • Do not: Mix sync and async clients. Avoid using httpx.Client inside async def functions, as it will block the event loop. Use httpx.AsyncClient instead.
  • Do not: Create clients in "hot loops". Move client instantiation outside of loops that perform many requests.

Dependency Injection ​

  • Do: Use Dependency Injector to manage dependencies.
  • Do: Define containers and providers in container.py.
  • Do: Use dependency injection to decouple components and invert control.

Why:

  • Flexibility: Components are loosely coupled. You can easily extend or change the functionality of a system by combining components in different ways.
  • Testability: Testing is easier because you can easily inject mocks instead of real objects that use APIs or databases.
  • Clearness and Maintainability: Dependency injection makes dependencies explicit. "Explicit is better than implicit" (PEP 20). It provides an overview and control of the application structure in one place.

Configuration Management ​

We use .env files for environment-specific configuration and a Pydantic-based AppConfig class from the backend_common package.

Full Documentation

For detailed documentation on configuration management, including how to use the built-in AppConfig or create custom configurations, see the backend-common Configuration Guide.

Quick summary:

  • Do: Place configuration in utils/app_config.py.
  • Do: Use get_env_or_throw() for required environment variables.
  • Do: Use os.getenv() with defaults for optional variables.
  • Do: Never commit secrets to version control; use .env files locally.
  • Do: Provide a .env.example with placeholder values.
  • Do not: Log sensitive values; use log_secret() to mask them.

Logging ​

We use structlog via backend_common.logger for structured, consistent logging across all services.

Implementation Guide

For setup instructions and API reference on using the backend_common.logger module, see the backend-common Logger Guide.

Log Levels ​

LevelWhen to Use
DEBUGDetailed diagnostic information (disabled in production)
INFOGeneral operational events (request received, task completed)
WARNINGUnexpected situations that don't prevent operation
ERRORErrors that prevent a specific operation from completing
CRITICALSystem-wide failures requiring immediate attention

Best Practices ​

  • Do: Use structured logging with key-value pairs, not string interpolation.
  • Do: Include contextual information (request IDs, user IDs, operation names).
  • Do: Log at appropriate levels (don't log expected errors as ERROR).
  • Do not: Log sensitive data (passwords, API keys, PII, tokens).
  • Do not: Log and re-raise exceptions (causes duplicate log entries).
python
# ✅ Right - Structured logging
logger.info("User authenticated", user_id=user.id, method="oauth")

# ❌ Wrong - String interpolation
logger.info(f"User {user.id} authenticated via oauth")

# ❌ Wrong - Logging sensitive data
logger.info("API call", api_key=config.api_key)  # NEVER DO THIS!

# ✅ Right - Mask secrets if needed for debugging
logger.debug("API configured", api_key=log_secret(config.api_key))

Folder Structure ​

Do: Organize projects using the following structure:

root/
 ├─ .github/             # Workflow and CI/CD
 ├─ src/
 │   └─ PROJECT_NAME/    # Source code directory
 │       ├─ __init__.py
 │       ├─ app.py
 │       ├─ container.py # Dependency injection container
 │       ├─ py.typed
 │       ├─ models/
 │       ├─ routers/
 │       ├─ services/
 │       └─ utils/
 ├─ tests/
 │   ├─ integration/     # Integration tests
 │   └─ unit/            # Unit tests
 ├─ scripts/             # Scripts for DB init, dataset prep, etc.
 ├─ .env
 ├─ .env.example
 ├─ .gitignore
 ├─ .pre-commit-config.yaml
 ├─ .python-version
 ├─ docker-compose.yml
 ├─ Dockerfile
 ├─ LICENSE
 ├─ Makefile
 ├─ pyproject.toml
 ├─ README.md
 └─ renovate.json

Dependency Management ​

  • Do: Use pyproject.toml for dependency management.
  • Do: Use dependency groups to separate development dependencies from production ones to keep Docker images small.
  • Do: Define scripts in [project.scripts] for tasks like data preparation or model training.

Dependency Cooldowns ​

To prevent instability, catch potential regressions, and mitigate supply chain attacks, we use dependency cooldowns. This ensures that we only update to new package versions after they have been available for a certain period, allowing time for malicious or buggy releases to be identified and retracted by the community.

We configure this using uv in pyproject.toml (see uv documentation):

toml
[tool.uv]
# Global cooldown: ignore releases newer than 1 week
exclude-newer = "1 week"

# Override for specific internal packages (e.g., 1 day cooldown)
exclude-newer-package = { dcc-backend-common = "P1D" }

Example pyproject.toml snippet for scripts:

toml
[project.scripts]
optimize-translation = "bs_translator_backend.scripts.optimize_llm:main"
prepare-dataset = "bs_translator_backend.scripts.prepare_dataset:main"

Data Structures ​

Choose the right data structure based on your use case:

StructureUse CaseValidationImmutable
Pydantic BaseModelAPI input/output, configuration, external data✅ YesOptional
dataclassInternal data containers, no validation needed❌ NoOptional
TypedDictDict-like data with type hints (JSON responses)❌ NoN/A

Pydantic Models ​

Use Pydantic for API request/response models and any data that needs validation:

python
from pydantic import BaseModel, Field, field_validator


class TranslationRequest(BaseModel):
    """Request payload for translation endpoint."""

    text: str = Field(..., min_length=1, max_length=10000, description="Text to translate")
    source_lang: str = Field(..., pattern=r"^[a-z]{2}$", description="ISO 639-1 source language")
    target_lang: str = Field(..., pattern=r"^[a-z]{2}$", description="ISO 639-1 target language")
    preserve_formatting: bool = Field(default=True, description="Maintain original formatting")

    @field_validator("target_lang")
    @classmethod
    def target_must_differ_from_source(cls, v: str, info) -> str:
        if v == info.data.get("source_lang"):
            raise ValueError("Target language must differ from source language")
        return v


class TranslationResponse(BaseModel):
    """Response payload for translation endpoint."""

    translated_text: str
    source_lang: str
    target_lang: str
    confidence: float = Field(..., ge=0.0, le=1.0)

Dataclasses ​

Use dataclasses for internal data structures that don't need validation:

python
from dataclasses import dataclass, field


@dataclass(frozen=True, slots=True)
class TranslationMetrics:
    """Internal metrics for translation operations."""

    tokens_used: int
    processing_time_ms: float
    cache_hit: bool = False


@dataclass(slots=True)
class TranslationContext:
    """Mutable context passed through translation pipeline."""

    request_id: str
    user_id: str
    translations: list[str] = field(default_factory=list)

    def add_translation(self, text: str) -> None:
        self.translations.append(text)

TypedDict ​

Use TypedDict for typing dict-like structures (e.g., JSON responses from external APIs):

python
from typing import TypedDict, NotRequired


class OpenAIMessage(TypedDict):
    """Structure of an OpenAI chat message."""

    role: str
    content: str


class OpenAIResponse(TypedDict):
    """Structure of an OpenAI API response."""

    id: str
    choices: list[dict]
    usage: NotRequired[dict]  # Optional field


def parse_openai_response(data: OpenAIResponse) -> str:
    """Extract content from OpenAI response."""
    return data["choices"][0]["message"]["content"]

API Design Patterns ​

Best practices for designing FastAPI endpoints.

Request/Response Naming ​

  • Do: Use descriptive names: CreateUserRequest, UserResponse, UpdateUserRequest.
  • Do: Keep request and response models separate (even if similar).
python
# models/user.py
from pydantic import BaseModel, EmailStr, Field


class CreateUserRequest(BaseModel):
    """Request to create a new user."""

    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)
    role: str = Field(default="user")


class UpdateUserRequest(BaseModel):
    """Request to update an existing user."""

    name: str | None = None
    role: str | None = None


class UserResponse(BaseModel):
    """User data returned by the API."""

    id: str
    email: str
    name: str
    role: str
    created_at: datetime

Router Organization ​

python
# routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status

from myapp.models.user import CreateUserRequest, UserResponse, UpdateUserRequest
from myapp.services.user_service import UserService

router = APIRouter(prefix="/users", tags=["users"])


@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    request: CreateUserRequest,
    user_service: UserService = Depends(),
) -> UserResponse:
    """Create a new user account."""
    return await user_service.create(request)


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: str,
    user_service: UserService = Depends(),
) -> UserResponse:
    """Retrieve a user by ID."""
    user = await user_service.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user


@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: str,
    request: UpdateUserRequest,
    user_service: UserService = Depends(),
) -> UserResponse:
    """Update an existing user."""
    return await user_service.update(user_id, request)


@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    user_id: str,
    user_service: UserService = Depends(),
) -> None:
    """Delete a user account."""
    await user_service.delete(user_id)

Error Responses ​

Use consistent error response format across all endpoints:

python
from pydantic import BaseModel


class ErrorResponse(BaseModel):
    """Standard error response format."""

    detail: str
    code: str | None = None
    field: str | None = None


# Usage in exception handlers
@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=ErrorResponse(
            detail=str(exc),
            code="VALIDATION_ERROR",
        ).model_dump(),
    )

HTTP Status Codes ​

OperationSuccessCommon Errors
POST (create)201 Created400, 409 Conflict
GET (read)200 OK404 Not Found
PATCH/PUT (update)200 OK404, 400
DELETE204 No Content404

Testing ​

We use pytest as our testing framework with native pytest assertions.

Test Organization ​

  • Do: Place tests in the tests/ directory.
  • Do: Separate unit tests (fast, run in CI/CD) from integration tests (slower, excluded from CI/CD, used locally).
  • Do: Run unit tests in the CI/CD pipeline.
  • Do: Mirror the source structure in tests (e.g., src/myapp/services/translator.py → tests/unit/services/test_translator.py).

Test Naming ​

  • Do: Name test files with test_ prefix: test_<module_name>.py
  • Do: Name test functions descriptively: test_<function>_<scenario>_<expected_result>
python
# ✅ Right - Descriptive test names
def test_translate_text_with_valid_input_returns_translation():
    ...

def test_translate_text_with_empty_string_raises_value_error():
    ...

def test_translate_text_with_unsupported_language_raises_translation_error():
    ...


# ❌ Wrong - Vague test names
def test_translate():
    ...

def test_error():
    ...

Writing Tests ​

  • Do: Use native assert statements (pytest provides detailed failure messages).
  • Do: Use pytest.raises for testing exceptions.
  • Do: Use pytest.mark.parametrize for testing multiple inputs.
  • Do: Write tests with minimal mocking; prefer real objects when possible.
  • Do: Use fixtures for shared setup.
python
import pytest
from myapp.services.translator import translate_text, TranslationError


# ✅ Right - Simple test with assert
def test_translate_text_returns_correct_translation():
    result = translate_text("Hello", source_lang="en", target_lang="de")

    assert result.text == "Hallo"
    assert result.source_lang == "en"
    assert result.target_lang == "de"


# ✅ Right - Testing exceptions
def test_translate_text_with_invalid_language_raises_error():
    with pytest.raises(ValueError, match="Invalid language code"):
        translate_text("Hello", source_lang="en", target_lang="invalid")


# ✅ Right - Parameterized tests for multiple scenarios
@pytest.mark.parametrize(
    ("input_text", "expected"),
    [
        ("Hello", "Hallo"),
        ("Goodbye", "Auf Wiedersehen"),
        ("Thank you", "Danke"),
    ],
)
def test_translate_text_handles_common_phrases(input_text: str, expected: str):
    result = translate_text(input_text, source_lang="en", target_lang="de")
    assert result.text == expected

Fixtures ​

  • Do: Use fixtures for reusable test setup.
  • Do: Define shared fixtures in conftest.py.
  • Do: Use factory fixtures for creating test data with variations.
python
# conftest.py
import pytest
from myapp.container import Container


@pytest.fixture
def app_config():
    """Provide test configuration."""
    return AppConfig(
        openai_api_key="test-key",
        llm_model="gpt-4o-mini",
        # ... other test values
    )


@pytest.fixture
def translation_service(app_config):
    """Provide a configured translation service."""
    container = Container()
    container.config.override(app_config)
    return container.translation_service()


# Factory fixture for flexible test data
@pytest.fixture
def make_translation_request():
    """Factory for creating translation requests with defaults."""

    def _make(
        text: str = "Hello",
        source_lang: str = "en",
        target_lang: str = "de",
    ) -> TranslationRequest:
        return TranslationRequest(
            text=text,
            source_lang=source_lang,
            target_lang=target_lang,
        )

    return _make


# Using the factory fixture
def test_translate_with_factory(translation_service, make_translation_request):
    request = make_translation_request(text="Good morning")
    result = translation_service.translate(request)
    assert result.text is not None

Async Tests ​

Use pytest-asyncio for testing async code:

python
import pytest


@pytest.mark.asyncio
async def test_async_translate_returns_result():
    result = await async_translate("Hello", target_lang="de")
    assert result.text == "Hallo"


@pytest.mark.asyncio
async def test_async_translate_with_timeout_raises_error():
    with pytest.raises(TimeoutError):
        await async_translate("Hello", target_lang="de", timeout=0.001)

Versioning ​

We use Semantic Versioning (SemVer) for all packages and services.

Format: MAJOR.MINOR.PATCH (e.g., 2.1.0)

ComponentWhen to Increment
MAJORBreaking changes (incompatible API changes)
MINORNew features (backwards-compatible)
PATCHBug fixes (backwards-compatible)
  • Do: Define version in pyproject.toml under [project].
  • Do: Tag releases in Git with v prefix (e.g., v2.1.0).
  • Do: Update version before merging to main/production.
toml
# pyproject.toml
[project]
name = "my-service"
version = "2.1.0"

Decorators ​

Decorators are useful for cross-cutting concerns like logging, timing, retries, and caching.

  • Do: Place reusable decorators in utils/decorators.py.
  • Do: Use functools.wraps to preserve function metadata.
  • Do: Keep decorators simple and single-purpose.
python
# utils/decorators.py
import functools
import time
from typing import Callable, ParamSpec, TypeVar

from backend_common.logger import get_logger

logger = get_logger(__name__)

P = ParamSpec("P")
R = TypeVar("R")


def log_execution_time(func: Callable[P, R]) -> Callable[P, R]:
    """Log the execution time of a function."""

    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        logger.info(f"{func.__name__} executed", duration_ms=elapsed * 1000)
        return result

    return wrapper


def retry(max_attempts: int = 3, delay: float = 1.0):
    """Retry a function on failure with exponential backoff."""

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            last_exception: Exception | None = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        time.sleep(delay * (2**attempt))
            raise last_exception  # type: ignore

        return wrapper

    return decorator


# Usage
@log_execution_time
@retry(max_attempts=3)
def fetch_remote_data(url: str) -> dict:
    ...

Context Managers ​

Use context managers for resource management (files, connections, locks, transactions).

  • Do: Place reusable context managers in utils/context_managers.py.
  • Do: Use contextlib.contextmanager for simple cases.
  • Do: Use contextlib.asynccontextmanager for async resources.
python
# utils/context_managers.py
from contextlib import contextmanager, asynccontextmanager
from typing import Generator
import time

from backend_common.logger import get_logger

logger = get_logger(__name__)


@contextmanager
def timed_operation(name: str) -> Generator[None, None, None]:
    """Log the duration of an operation."""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        logger.info(f"{name} completed", duration_ms=elapsed * 1000)


@contextmanager
def temporary_env_var(key: str, value: str) -> Generator[None, None, None]:
    """Temporarily set an environment variable."""
    import os

    original = os.environ.get(key)
    os.environ[key] = value
    try:
        yield
    finally:
        if original is None:
            del os.environ[key]
        else:
            os.environ[key] = original


# Usage
with timed_operation("translation"):
    result = translate(text)

with temporary_env_var("DEBUG", "true"):
    run_debug_operation()

Constants & Enums ​

  • Do: Place module-level constants at the top of the file, after imports.
  • Do: Place shared constants in utils/constants.py.
  • Do: Use Enum or StrEnum for related constants with type safety.
python
# utils/constants.py
from enum import StrEnum


# Simple constants
DEFAULT_TIMEOUT_SECONDS = 30
MAX_RETRIES = 3
SUPPORTED_LANGUAGES = frozenset({"en", "de", "fr", "es", "it"})


# StrEnum for string constants (Python 3.11+)
class TranslationStatus(StrEnum):
    """Status of a translation request."""

    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"


class Language(StrEnum):
    """Supported languages."""

    ENGLISH = "en"
    GERMAN = "de"
    FRENCH = "fr"
    SPANISH = "es"
    ITALIAN = "it"


# Usage
def get_translation(request_id: str) -> TranslationResponse:
    status = get_status(request_id)
    if status == TranslationStatus.COMPLETED:
        return fetch_result(request_id)
    raise TranslationPendingError(status)

Where to place things:

ItemLocation
Module-specific constantsTop of the module file
Shared constantsutils/constants.py
Shared enumsutils/constants.py or models/enums.py
Decoratorsutils/decorators.py
Context managersutils/context_managers.py
Protocols/Interfacesinterfaces/ or alongside related code

Developed with ❤️ by the DCC. Documentation released under the MIT License.