service

service

REST API service for read-only database exposure via FastAPI.

Auto-generates paginated endpoints for all discovered tables, views, and materialized views. Per-table access control via TableConfig allows disabling individual endpoints.

The HTTP server runs as a background asyncio.Task alongside the standard run_forever() cycle. Each run() cycle logs request statistics and updates Prometheus metrics.

Note

Rate limiting is not enforced at the application level — it is expected to be handled by the reverse proxy (e.g., Cloudflare, Nginx). The API is strictly read-only: only GET methods are registered, and all queries are executed through the Catalog safe query builder.

See Also

ApiConfig: Configuration model for HTTP settings, pagination, and CORS. Catalog: Schema introspection and query builder shared with the DVM service. BaseService: Abstract base class providing lifecycle and metrics.

Examples:

from bigbrotr.core import Brotr
from bigbrotr.services import Api

brotr = Brotr.from_yaml("config/brotr.yaml")
api = Api.from_yaml("config/services/api.yaml", brotr=brotr)

async with brotr:
    async with api:
        await api.run_forever()

Classes

Api

Api(brotr: Brotr, config: ApiConfig | None = None)

Bases: CatalogAccessMixin, BaseService[ApiConfig]

REST API service exposing the BigBrotr database read-only.

Auto-generates paginated GET endpoints for each enabled table, view, and materialized view discovered by the Catalog.

Lifecycle

1. __aenter__: discover schema, build FastAPI app, start uvicorn.
2. run(): log statistics and update Prometheus gauges.
3. __aexit__: cancel the HTTP server task.

See Also

ApiConfig: Configuration model for this service. Dvm: Sibling service that exposes the same Catalog data via Nostr NIP-90.

Source code in src/bigbrotr/services/api/service.py

def __init__(self, brotr: Brotr, config: ApiConfig | None = None) -> None:
    super().__init__(brotr=brotr, config=config)
    self._config: ApiConfig
    self._server_task: asyncio.Task[None] | None = None
    self._requests_total = 0
    self._requests_failed = 0

Functions

cleanup async

cleanup() -> int

No-op: Api does not use service state.

Source code in src/bigbrotr/services/api/service.py

async def cleanup(self) -> int:
    """No-op: Api does not use service state."""
    return 0

run async

run() -> None

Log request stats and update Prometheus counters.

Detects a crashed HTTP server task and raises RuntimeError to trigger the run_forever() failure counter. Per-cycle request counters are snapshotted and reset atomically.

Source code in src/bigbrotr/services/api/service.py

async def run(self) -> None:
    """Log request stats and update Prometheus counters.

    Detects a crashed HTTP server task and raises ``RuntimeError``
    to trigger the ``run_forever()`` failure counter.  Per-cycle
    request counters are snapshotted and reset atomically.
    """
    if self._server_task is not None and self._server_task.done():
        exc = self._server_task.exception() if not self._server_task.cancelled() else None
        self._logger.error("http_server_crashed", error=str(exc) if exc else "cancelled")
        raise RuntimeError("HTTP server task has stopped unexpectedly") from exc

    # Snapshot and reset per-cycle counters
    total = self._requests_total
    failed = self._requests_failed
    self._requests_total = 0
    self._requests_failed = 0

    tables_exposed = sum(1 for name in self._catalog.tables if self._is_table_enabled(name))
    self._logger.info(
        "cycle_stats",
        requests_total=total,
        requests_failed=failed,
        tables_exposed=tables_exposed,
    )
    self.inc_counter("requests_total", total)
    self.inc_counter("requests_failed", failed)
    self.set_gauge("tables_exposed", tables_exposed)