Skip to content

framework-modules.md

Latest commit

 

History

History
582 lines (395 loc) · 18 KB

framework-modules.md

File metadata and controls

582 lines (395 loc) · 18 KB

Framework modules reference

Public API of the nene2 package.

nene2.http

PaginationQueryParser

Parses limit and offset query parameters.

FastAPI Depends (recommended):

from typing import Annotated
from fastapi import Depends
from nene2.http import PaginationQueryParser

@router.get("/items")
def list_items(pagination: Annotated[PaginationQueryParser, Depends()]) -> JSONResponse:
    result = use_case.execute(pagination.limit, pagination.offset)

Legacy (Request-based):

from nene2.http import PaginationQueryParser

pagination = PaginationQueryParser.parse(request)
# pagination.limit  → int (max 100, default 20)
# pagination.offset → int (default 0)

PaginationResponse

Wraps a paginated result set.

from nene2.http import PaginationResponse

body = PaginationResponse(items=[...], limit=20, offset=0, total=42).to_dict()
# → {"items": [...], "limit": 20, "offset": 0, "total": 42}

problem_details_response()

Generates an RFC 9457 Problem Details response.

from nene2.http import problem_details_response

return problem_details_response("not-found", "Not Found", 404, "Note 42 not found.")

PaginationQuery

Dataclass returned by PaginationQueryParser.parse(). Contains limit: int and offset: int.

HealthCheckProtocol / HealthStatus

Contract and result type for application health checks.

from nene2.http import HealthCheckProtocol, HealthStatus

class MyHealthCheck:
    def check(self) -> HealthStatus:
        return HealthStatus(status="ok")

HealthStatus fields: status: str ("ok" or "error"), checks: dict[str, str]. is_healthy property returns True when status == "ok".

ETag and conditional requests

from nene2.http import check_not_modified, check_precondition, generate_etag

etag = generate_etag({"id": 1, "title": "Hello"})
# Returns 304 when If-None-Match matches (GET)
check_not_modified(request, etag)
# Returns 412 when If-Match does not match (PUT/PATCH/DELETE)
check_precondition(request, etag)

Query parameter helpers

Typed parsers for common query patterns (raise ValidationException on invalid input):

from nene2.http import query_array, query_bool, query_comma_separated, query_int, query_string

limit = query_int(request, "limit", default=20, minimum=1, maximum=100)
tags = query_comma_separated(request, "tags", max_items=10)

RequestScopedContext[T]

Request-scoped value holder for dependency injection (see lifespan-and-app-state).

PaginationDep

FastAPI Depends() alias for PaginationQueryParser — preferred over manual parsing.

nene2.use_case

UseCaseProtocol[I, O]

Structural contract for synchronous UseCases.

from nene2.use_case import UseCaseProtocol

class MyUseCase:
    def execute(self, input_: MyInput) -> MyOutput: ...

assert isinstance(MyUseCase(), UseCaseProtocol)

AsyncUseCaseProtocol[I, O]

Structural contract for async UseCases.

from nene2.use_case import AsyncUseCaseProtocol

class MyAsyncUseCase:
    async def execute(self, input_: MyInput) -> MyOutput: ...

assert isinstance(MyAsyncUseCase(), AsyncUseCaseProtocol)

Note: isinstance checks attribute presence only. The async/sync distinction is enforced statically by mypy --strict.

nene2.config

AppSettings

Pydantic Settings class — reads from environment variables and .env.

from nene2.config import AppSettings

cfg = AppSettings()                                   # from environment
cfg_test = AppSettings(throttle_enabled=False)        # override for tests

See Configuration reference for all fields.

nene2.middleware

ErrorHandlerMiddleware

Catches all unhandled exceptions and converts them to Problem Details responses. Register domain exception handlers via DomainExceptionHandlerProtocol.

from starlette.responses import Response
from nene2.http import problem_details_response
from nene2.middleware import ErrorHandlerMiddleware
from nene2.middleware.domain_exception import DomainExceptionHandlerProtocol

class NoteNotFoundExceptionHandler:
    def handles(self, exc: Exception) -> bool:
        return isinstance(exc, NoteNotFoundException)

    def handle(self, exc: Exception) -> Response:
        assert isinstance(exc, NoteNotFoundException)
        return problem_details_response("not-found", "Not Found", 404, str(exc))

# Registration — pass as domain_handlers list:
app.add_middleware(
    ErrorHandlerMiddleware,
    debug=settings.app_debug,
    domain_handlers=[NoteNotFoundExceptionHandler()],
)

DomainExceptionHandlerProtocol requires two methods:

Method Signature Purpose
handles (exc: Exception) -> bool Return True if this handler owns the exception
handle (exc: Exception) -> Response Convert exception to an HTTP response

Other middleware

Class Module Role
SecurityHeadersMiddleware nene2.middleware.security_headers Add security response headers
RequestIdMiddleware nene2.middleware.request_id Generate / propagate X-Request-ID
RequestLoggingMiddleware nene2.middleware.request_logging Structured request / response logging
RequestSizeLimitMiddleware nene2.middleware.request_size_limit Reject oversized request bodies
ThrottleMiddleware nene2.middleware.throttle Fixed-window rate limiting per IP

add_middleware arguments

Starlette applies middleware in reverse registration order — the last registered becomes the outermost layer. Register ErrorHandlerMiddleware first so it catches all exceptions from every other middleware.

Middleware Keyword arguments Default
ErrorHandlerMiddleware debug: bool, domain_handlers: list[DomainExceptionHandlerProtocol] | None False, None
SecurityHeadersMiddleware (none) —
RequestIdMiddleware (none) —
RequestLoggingMiddleware (none) —
RequestSizeLimitMiddleware max_bytes: int 1_048_576 (1 MiB)
ThrottleMiddleware limit: int, window: int 60, 60

ThrottleMiddleware has no enabled flag — wrap with if settings.throttle_enabled: to disable it.

Note — X-Forwarded-For spoofing: The rate limit key is derived from the first entry of the X-Forwarded-For header, which clients can forge. In production, always place the application behind a trusted reverse proxy (nginx, Caddy, AWS ALB, etc.) that rewrites X-Forwarded-For before the request reaches the app. See ADR-0006 for details.

Full registration order with optional middleware

# Registration order: innermost first, outermost last.
# Starlette executes in reverse — the last registered wraps all others.
app.add_middleware(ErrorHandlerMiddleware, debug=settings.app_debug, domain_handlers=[...])
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(RequestIdMiddleware)
app.add_middleware(RequestLoggingMiddleware)
app.add_middleware(RequestSizeLimitMiddleware, max_bytes=settings.max_body_size)
if settings.throttle_enabled:
    app.add_middleware(ThrottleMiddleware, limit=settings.throttle_limit, window=settings.throttle_window)
# Auth middleware — registered before CORS so it sits inside the CORS layer
if settings.bearer_token_enabled:
    app.add_middleware(BearerTokenMiddleware, verifier=LocalTokenVerifier(settings.bearer_tokens))
if settings.api_key_enabled:
    app.add_middleware(ApiKeyAuthMiddleware, verifier=LocalTokenVerifier(settings.api_keys))
# CORS must be the outermost layer — register it last.
# OPTIONS preflight requests must reach CORSMiddleware before any auth check.
# If CORSMiddleware is registered before auth middleware, the auth layer becomes
# outermost and returns 401 on preflight, breaking CORS for all browsers.
if settings.cors_enabled:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=settings.cors_allow_credentials,
        allow_methods=settings.cors_allow_methods,
        allow_headers=settings.cors_allow_headers,
    )

CORS + Auth rule: Always register CORSMiddleware after any auth middleware. In Starlette's reverse order, "last registered = outermost" means CORS wraps auth, so browser preflight (OPTIONS) requests are handled before authentication.

setup_middlewares()

Registers the full nene2 middleware stack in the correct LIFO order (including optional CORS). Prefer this over manual add_middleware calls when you do not need custom middleware.

from nene2.middleware import setup_middlewares

setup_middlewares(
    app,
    debug=settings.app_debug,
    domain_handlers=[NoteNotFoundExceptionHandler()],
    throttle_limit=settings.throttle_limit if settings.throttle_enabled else None,
    max_request_bytes=settings.max_body_size,
    cors_allowed_origins=settings.cors_origins if settings.cors_enabled else None,
)

See middleware-stack how-to.

SimpleDomainHandler

Helper to build DomainExceptionHandlerProtocol from an exception type and status code.

Rate limit storage

Symbol Role
RateLimitStorageProtocol Pluggable storage for throttle counters
InMemoryRateLimitStorage Default in-process implementation
ThrottleMiddleware Accepts optional storage= for custom backends

nene2.auth

LocalTokenVerifier

Verifies tokens against a static list using secrets.compare_digest.

from nene2.auth import LocalTokenVerifier

verifier = LocalTokenVerifier(["token-a", "token-b"])
verifier.verify("token-a")  # True
verifier.verify("wrong")    # False

TokenVerifierProtocol / TokenIssuerProtocol

Structural contracts for custom verifiers and issuers (e.g. JWT).

TokenVerificationException

Raise this from a verifier to signal an invalid token. BearerTokenMiddleware maps it to 401 Unauthorized.

CompositeAuthMiddleware

Path-prefix rules for mixed auth (e.g. Bearer on /api/*, API Key on /internal/*).

from nene2.auth import CompositeAuthMiddleware, CompositeAuthRule, bearer_check, api_key_check

app.add_middleware(
    CompositeAuthMiddleware,
    rules=[
        CompositeAuthRule(prefix="/api", check=bearer_check(verifier)),
        CompositeAuthRule(prefix="/internal", check=api_key_check(verifier)),
    ],
)

LocalTokenIssuer / LocalBearerJwtVerifier

Development helpers for HMAC-signed bearer tokens (see src/example/ protected routes).

make_require_auth()

FastAPI Depends() factory that returns 401 Problem Details when auth headers are missing.

nene2.database

SqlAlchemyQueryExecutor

Executes parameterised SQL via SQLAlchemy Core.

from nene2.database import SqlAlchemyQueryExecutor

executor = SqlAlchemyQueryExecutor(engine)
rows = executor.fetch_all("SELECT * FROM notes WHERE id = :id", {"id": 1})
executor.write("INSERT INTO notes (title, body) VALUES (:t, :b)", {"t": "t", "b": "b"})

write() return value

write() returns an int whose meaning depends on the SQL operation:

Operation Return value
INSERT with AUTOINCREMENT / SERIAL lastrowid — the new row's primary key (always > 0)
INSERT without auto-PK, or multi-row INSERT rowcount — number of rows inserted
UPDATE / DELETE rowcount — rows affected (0 if nothing matched)

Use lastrowid to reconstruct the entity after a single-row INSERT:

new_id = executor.write("INSERT INTO notes (title) VALUES (:title)", {"title": "Hello"})
return Note(id=new_id, title="Hello")

Use rowcount to detect a missing row on UPDATE / DELETE:

affected = executor.write("UPDATE notes SET title=:title WHERE id=:id", {"title": t, "id": pk})
if affected == 0:
    raise NoteNotFoundException(pk)

SqlAlchemyTransactionManager

Manages transactions. Prefer transactional() over manual begin/commit/rollback.

from nene2.database import SqlAlchemyTransactionManager

mgr = SqlAlchemyTransactionManager(engine)

result = mgr.transactional(
    lambda ex: ex.fetch_one("SELECT COUNT(*) AS cnt FROM notes")
)

Combining transactional() with the Repository pattern

When a UseCase needs to perform multiple writes atomically, define _in_tx variants on the repository interface that accept an explicit executor. The UseCase passes the transaction-bound executor from the callback to each _in_tx method.

Repository interface:

from nene2.database import DatabaseQueryExecutorInterface
from abc import ABC, abstractmethod

class AccountRepositoryInterface(ABC):
    # Standard methods — use self._executor (auto-commit)
    @abstractmethod
    def find_by_id(self, account_id: int) -> Account | None: ...

    # _in_tx variants — call only inside a transactional() callback
    @abstractmethod
    def find_by_id_in_tx(
        self, executor: DatabaseQueryExecutorInterface, account_id: int
    ) -> Account | None: ...

    @abstractmethod
    def update_balance_in_tx(
        self, executor: DatabaseQueryExecutorInterface, account_id: int, delta: int
    ) -> None: ...

UseCase (atomic transfer example):

from nene2.database import DatabaseQueryExecutorInterface, DatabaseTransactionManagerInterface

class TransferUseCase:
    def __init__(
        self,
        transaction_manager: DatabaseTransactionManagerInterface,
        account_repo: AccountRepositoryInterface,
        transfer_repo: TransferRepositoryInterface,
    ) -> None:
        self._tx = transaction_manager
        self._accounts = account_repo
        self._transfers = transfer_repo

    def execute(self, input_: TransferInput) -> Transfer:
        def _run(executor: DatabaseQueryExecutorInterface) -> Transfer:
            source = self._accounts.find_by_id_in_tx(executor, input_.from_account_id)
            if source is None:
                raise AccountNotFoundException(input_.from_account_id)
            if source.balance_cents < input_.amount_cents:
                raise InsufficientBalanceException(...)

            self._accounts.update_balance_in_tx(executor, input_.from_account_id, -input_.amount_cents)
            self._accounts.update_balance_in_tx(executor, input_.to_account_id, input_.amount_cents)
            return self._transfers.create_in_tx(executor, input_.from_account_id, input_.to_account_id, input_.amount_cents)

        return self._tx.transactional(_run)

transactional() uses engine.begin() internally — any exception inside the callback triggers an automatic rollback.

Testing with InMemory: Implement DatabaseTransactionManagerInterface with a no-op executor that calls the callback directly. The _in_tx methods on the InMemory repository ignore the executor and operate on their in-memory store.

DatabaseHealthCheck

Implements HealthCheckProtocol — verifies the database connection and returns a HealthStatus.

from nene2.database import DatabaseHealthCheck
from nene2.http import HealthStatus

health = DatabaseHealthCheck(engine)
status: HealthStatus = health.check()
# status.status → "ok" or "error"
# status.checks → {"db": "ok"} or {"db": "error: <message>"}

DatabaseConnectionException

Raised by DatabaseHealthCheck or repository operations when the database is unreachable.

nene2.mcp

LocalMcpServer

Wraps FastMCP — registers UseCase functions as MCP tools.

from nene2.mcp import LocalMcpServer

server = LocalMcpServer("my-server", instructions="...")

@server.tool("List all notes.")
def list_notes(limit: int = 20, offset: int = 0) -> list[dict]: ...

server.run(transport="stdio")

HttpxMcpClient

HTTP client for calling a nene2 API from MCP tool handlers.

from nene2.mcp import HttpxMcpClient

client = HttpxMcpClient("bearer-token")
response = client.get("http://localhost:8080", "/notes")
response.is_successful()   # True
response.body              # str — raw response text
response.status_code       # int
response.request_id()      # str | None — value of X-Request-ID header

McpHttpResponse

Return type of HttpxMcpClient methods.

Fields: status_code: int, headers: dict[str, str], body: str (raw response text).

Methods:

  • is_successful() -> bool — True when 200 ≤ status_code < 300
  • request_id() -> str | None — returns the X-Request-ID response header value, or None

McpHttpClientProtocol

Structural contract for custom MCP HTTP clients. Implement get(), post(), put(), delete() returning McpHttpResponse, and has_authentication() -> bool.

nene2.log

setup_logging()

Initialises structlog. Switches between ConsoleRenderer (local) and JSON (production).

from nene2.log import setup_logging

setup_logging(app_env="production")  # JSON renderer
setup_logging(app_env="local")       # Console renderer

nene2.validation

ValidationException / ValidationError

Raise ValidationException at the HTTP boundary to return 422 Unprocessable Entity.

from nene2.validation.exceptions import ValidationError, ValidationException

errors = [ValidationError("body", "Body must not be empty.", "required")]
raise ValidationException(errors)

nene2.cache

TtlCache[V]

Thread-safe in-memory cache with per-key TTL expiry. Use for idempotency keys, short-lived lookups, or rate-limit adjuncts.

from nene2.cache import TtlCache

cache: TtlCache[str] = TtlCache(ttl_seconds=60.0)
cache.set("key", "value")
cache.get("key")  # str | None

See lifespan-and-app-state how-to for app.state wiring.

nene2.security

verify_hmac_signature()

Timing-safe HMAC verification for webhook endpoints.

from nene2.security import verify_hmac_signature

if not verify_hmac_signature(body, signature_header, secret.get_secret_value()):
    return problem_details_response("unauthorized", "Unauthorized", 401, "Invalid signature.")

See webhook how-to.