diff --git a/AGENTS.md b/AGENTS.md index 719f8a1..56416f3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -35,7 +35,7 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f 2. Builds a `FastAPIBootstrapper` from `settings.api_bootstrapper_config`, injecting the SQLAlchemy OpenTelemetry instrumentor. 3. `modern_di_fastapi.setup_di(app, container)` wires DI scopes onto the FastAPI app. 4. Includes one `ROUTER` per resource (`app.api.decks.ROUTER`, `app.api.cards.ROUTER`) under `/api` via `include_routers`. Add a new resource by creating `app/api/.py` with its own `APIRouter`-based `ROUTER` and registering it there. -5. Registers `DuplicateKeyError` → 422 handler from `app/exceptions.py`. +5. Registers handlers from `app/exceptions.py`: `DuplicateKeyError` → 422, `NotFoundError` and `ForeignKeyError` → 404, and `sqlalchemy.exc.TimeoutError` and `RepositoryError` → `database_unavailable_handler`, which returns 503 for a pool timeout and re-raises anything else. ### DI scopes (modern-di) @@ -55,6 +55,10 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f Endpoints inject a use case with `FromDI(UseCase)` from `modern_di_fastapi`, call it, and convert the result to a schema. They never touch repositories or the session. Add new providers to the matching group rather than constructing services manually in routes. +### Pool and server sizing + +`DB_POOL_SIZE` (positive) and `DB_MAX_OVERFLOW` (non-negative) size the pool, and `app/__main__.py` sets Granian's `backpressure` to their sum, so each worker accepts only as many requests at once as it has connections. Requests beyond that wait in Granian's socket backlog instead of queueing on the pool. `APP_WORKERS` sets the Granian worker count, and each worker has its own pool. `DB_POOL_TIMEOUT` (default 30 s) bounds the wait for a connection, and a timeout returns 503 through `exceptions.database_unavailable_handler`. advanced-alchemy wraps the timeout in a `RepositoryError`, so the handler is also registered for `RepositoryError` and re-raises anything not caused by a pool timeout. `DB_POOL_RECYCLE` defaults to SQLAlchemy's `-1` (never). Set it below the idle timeout of any proxy or load balancer in front of Postgres, so a connection it dropped is not handed out and retried after a 1-2 s backoff. + ### Use cases and lost connections `pool_pre_ping` is off by default (`DB_POOL_PRE_PING`). A pooled connection that died while idle is handled by `db_retry.postgres_retry` instead, which saves a round trip on every checkout. That only works if the retried function owns its whole transaction, so every use case has this shape: @@ -91,7 +95,7 @@ class ListDecksUseCase: ### Settings -`app/settings.py` — `pydantic_settings.BaseSettings`. Env vars are unprefixed (`DB_DSN`, `DB_REPLICA_DSN`, `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `LOG_LEVEL`, `APP_HOST`, `APP_PORT`, `OPENTELEMETRY_ENDPOINT`, `SENTRY_DSN`, `CORS_ALLOWED_ORIGINS`, ...). `api_bootstrapper_config` produces a `FastAPIConfig` for `lite-bootstrap`. `LOG_LEVEL` is case-insensitive and sets both Granian's log level and the app's root logger. +`app/settings.py` — `pydantic_settings.BaseSettings`. Env vars are unprefixed (`DB_DSN`, `DB_REPLICA_DSN`, `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `LOG_LEVEL`, `APP_HOST`, `APP_PORT`, `OPENTELEMETRY_ENDPOINT`, `SENTRY_DSN`, `CORS_ALLOWED_ORIGINS`, `APP_WORKERS`, `DB_POOL_TIMEOUT`, `DB_POOL_RECYCLE`, ...). `api_bootstrapper_config` produces a `FastAPIConfig` for `lite-bootstrap`. `LOG_LEVEL` is case-insensitive and sets both Granian's log level and the app's root logger. ### Tests diff --git a/app/__main__.py b/app/__main__.py index c25b31d..386d0e9 100644 --- a/app/__main__.py +++ b/app/__main__.py @@ -10,6 +10,8 @@ factory=True, address=settings.app_host, port=settings.app_port, + workers=settings.app_workers, + backpressure=settings.db_pool_size + settings.db_max_overflow, interface=Interfaces.ASGI, log_level=settings.log_level, loop=Loops.uvloop, diff --git a/app/application.py b/app/application.py index 91ba52d..33fe3ff 100644 --- a/app/application.py +++ b/app/application.py @@ -3,9 +3,10 @@ import modern_di import modern_di_fastapi -from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError +from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError, RepositoryError from lite_bootstrap import FastAPIBootstrapper from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor +from sqlalchemy import exc as sa_exc from app import exceptions, ioc from app.api import cards, decks @@ -43,4 +44,6 @@ def build_app() -> fastapi.FastAPI: ForeignKeyError, exceptions.not_found_error_handler, # ty: ignore[invalid-argument-type] ) + app.add_exception_handler(sa_exc.TimeoutError, exceptions.database_unavailable_handler) + app.add_exception_handler(RepositoryError, exceptions.database_unavailable_handler) return app diff --git a/app/exceptions.py b/app/exceptions.py index ba364f0..d37f861 100644 --- a/app/exceptions.py +++ b/app/exceptions.py @@ -1,6 +1,7 @@ from typing import TYPE_CHECKING from fastapi.responses import JSONResponse +from sqlalchemy import exc as sa_exc from starlette import status @@ -21,3 +22,12 @@ async def not_found_error_handler(_: Request, __: NotFoundError | ForeignKeyErro status_code=status.HTTP_404_NOT_FOUND, content={"detail": "Not found"}, ) + + +async def database_unavailable_handler(_: Request, exc: Exception) -> JSONResponse: + if not isinstance(exc, sa_exc.TimeoutError) and not isinstance(exc.__cause__, sa_exc.TimeoutError): + raise exc + return JSONResponse( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + content={"detail": "Database unavailable"}, + ) diff --git a/app/resources/db.py b/app/resources/db.py index 8747e32..3e39f7a 100644 --- a/app/resources/db.py +++ b/app/resources/db.py @@ -24,6 +24,8 @@ def create_sa_engine(url: URL, *, read_only: bool = False) -> sa.AsyncEngine: pool_size=settings.db_pool_size, pool_pre_ping=settings.db_pool_pre_ping, max_overflow=settings.db_max_overflow, + pool_timeout=settings.db_pool_timeout, + pool_recycle=settings.db_pool_recycle, isolation_level="AUTOCOMMIT" if read_only else None, connect_args={"server_settings": {"default_transaction_read_only": "on"}} if read_only else {}, ) diff --git a/app/settings.py b/app/settings.py index a6d0d93..0b69f56 100644 --- a/app/settings.py +++ b/app/settings.py @@ -16,12 +16,15 @@ class Settings(pydantic_settings.BaseSettings): db_dsn: str = "postgresql+asyncpg://postgres:password@db/postgres" db_replica_dsn: str = "" - db_pool_size: int = 5 - db_max_overflow: int = 0 + db_pool_size: pydantic.PositiveInt = 5 + db_max_overflow: pydantic.NonNegativeInt = 0 db_pool_pre_ping: bool = False + db_pool_timeout: float = 30 + db_pool_recycle: int = -1 app_host: str = "0.0.0.0" # noqa: S104 app_port: int = 8000 + app_workers: int = 1 opentelemetry_endpoint: str = "" sentry_dsn: str = "" diff --git a/tests/test_main.py b/tests/test_main.py index 1a9ecd3..6a47d5a 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -6,6 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app import ioc +from app.settings import settings if TYPE_CHECKING: @@ -13,9 +14,17 @@ def test_main(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr("granian.Granian", mock.Mock()) + granian = mock.Mock() + monkeypatch.setattr("granian.Granian", granian) + monkeypatch.setattr(settings, "app_workers", 4) + monkeypatch.setattr(settings, "db_pool_size", 5) + monkeypatch.setattr(settings, "db_max_overflow", 2) + runpy.run_module("app.__main__", run_name="__main__") + assert granian.call_args.kwargs["workers"] == 4 # noqa: PLR2004 + assert granian.call_args.kwargs["backpressure"] == 7 # noqa: PLR2004 + async def test_session() -> None: container = modern_di.Container(groups=ioc.ALL_GROUPS) diff --git a/tests/test_pool.py b/tests/test_pool.py new file mode 100644 index 0000000..97126b2 --- /dev/null +++ b/tests/test_pool.py @@ -0,0 +1,52 @@ +import typing +import uuid + +import pytest +from advanced_alchemy.exceptions import RepositoryError +from fastapi import status + +from app import ioc +from app.resources.db import create_primary_sa_engine +from app.settings import settings +from tests import factories + + +if typing.TYPE_CHECKING: + import modern_di + from httpx import AsyncClient + + +@pytest.mark.parametrize( + "request_kwargs", + [ + {"method": "get", "url": "/api/decks/1/"}, + {"method": "post", "url": "/api/decks/", "json": {"name": "pool", "idempotency_key": str(uuid.uuid4())}}, + ], +) +async def test_exhausted_pool_returns_503( + client: AsyncClient, + di_container: modern_di.Container, + monkeypatch: pytest.MonkeyPatch, + request_kwargs: dict[str, typing.Any], +) -> None: + monkeypatch.setattr(settings, "db_pool_size", 1) + monkeypatch.setattr(settings, "db_max_overflow", 0) + monkeypatch.setattr(settings, "db_pool_timeout", 0.1) + engine: typing.Final = create_primary_sa_engine() + di_container.override(ioc.Database.dynamic_engine, engine) + try: + async with engine.connect(): + response = await client.request(**request_kwargs) + finally: + di_container.reset_override(ioc.Database.dynamic_engine) + await engine.dispose() + + assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE + + +@pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory") +async def test_other_repository_errors_are_not_turned_into_503(client: AsyncClient) -> None: + deck: typing.Final = await factories.DeckModelFactory.create_async() + + with pytest.raises(RepositoryError): + await client.post(f"/api/decks/{deck.id}/cards/", json=[{"front": "nul \x00 byte"}]) diff --git a/tests/test_settings.py b/tests/test_settings.py index 149dac8..83ae543 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -1,5 +1,7 @@ import logging +import pydantic +import pytest from granian.log import LogLevels from app.settings import Settings @@ -23,3 +25,9 @@ def test_log_level_is_case_insensitive_and_reaches_app_logging() -> None: assert custom.log_level == LogLevels.warning assert custom.api_bootstrapper_config.logging_log_level == logging.WARNING + + +@pytest.mark.parametrize("pool_settings", [{"db_pool_size": 0}, {"db_max_overflow": -1}]) +def test_pool_settings_reject_unbounded_pools(pool_settings: dict[str, int]) -> None: + with pytest.raises(pydantic.ValidationError): + Settings.model_validate(pool_settings)