Skip to content

API Schemas and Settings

The request and response shapes the REST API speaks, and the settings that configure a running instance.

API Schemas

orionbelt.api.schemas

API request/response Pydantic schemas.

SessionCreateRequest

Bases: BaseModel

Request body for POST /sessions.

Source code in src/orionbelt/api/schemas.py
class SessionCreateRequest(BaseModel):
    """Request body for POST /sessions."""

    metadata: dict[str, str] = Field(default_factory=dict)

SessionResponse

Bases: BaseModel

Single session info.

Source code in src/orionbelt/api/schemas.py
class SessionResponse(BaseModel):
    """Single session info."""

    session_id: str
    created_at: datetime
    last_accessed_at: datetime
    model_count: int
    metadata: dict[str, str] = Field(default_factory=dict)
    expires_at: datetime = Field(description="Idle TTL deadline (refreshed on each access)")
    max_expires_at: datetime = Field(description="Absolute lifetime deadline (fixed at creation)")

SessionListResponse

Bases: BaseModel

Response for GET /sessions.

Source code in src/orionbelt/api/schemas.py
class SessionListResponse(BaseModel):
    """Response for GET /sessions."""

    sessions: list[SessionResponse]

ModelLoadRequest

Bases: BaseModel

Request body for POST /sessions/{session_id}/models.

Source code in src/orionbelt/api/schemas.py
class ModelLoadRequest(BaseModel):
    """Request body for POST /sessions/{session_id}/models."""

    model_yaml: str | None = Field(
        default=None,
        description="OBML model as YAML string (provide model_yaml OR model_json)",
        max_length=5_000_000,
    )
    model_json: dict[str, object] | str | None = Field(
        default=None,
        description="OBML model as JSON object or JSON string (auto-parsed)",
    )
    extends: list[str] | None = Field(
        default=None,
        description="Optional inline YAML strings of analytical fragments to merge",
    )
    inherits: str | None = Field(
        default=None,
        description="Optional model ID of an already-loaded parent model in the session",
    )
    dedup: bool = Field(
        default=True,
        description=(
            "When True (default), identical OBML content already loaded in this session "
            "reuses the existing model_id (response.model_load == 'reused'). "
            "When False, always loads fresh."
        ),
    )

    @model_validator(mode="after")
    def _parse_model_json_string(self) -> ModelLoadRequest:
        if isinstance(self.model_json, str):
            self.model_json = json.loads(self.model_json)
        return self

ModelLoadResponse

Bases: BaseModel

Response for POST /sessions/{session_id}/models.

Source code in src/orionbelt/api/schemas.py
class ModelLoadResponse(BaseModel):
    """Response for POST /sessions/{session_id}/models."""

    model_id: str
    data_objects: int
    dimensions: int
    measures: int
    metrics: int
    warnings: list[StructuredWarning] = Field(default_factory=list)
    model_load: str = Field(
        default="fresh",
        description=(
            "Whether the load parsed a fresh model or reused an existing one. "
            "Values: 'fresh' | 'reused'."
        ),
    )
    health: ModelHealth | None = Field(
        default=None,
        description=(
            "Structural health of the model's join graph: orphan dataObjects, "
            "fan-trap risks, unreachable dimensions. Always present on a fresh load."
        ),
    )

ModelSummaryResponse

Bases: BaseModel

Short model summary for listing.

Source code in src/orionbelt/api/schemas.py
class ModelSummaryResponse(BaseModel):
    """Short model summary for listing."""

    model_id: str
    data_objects: int
    dimensions: int
    measures: int
    metrics: int

SessionQueryRequest

Bases: BaseModel

Request body for POST /sessions/{session_id}/query/sql.

Source code in src/orionbelt/api/schemas.py
class SessionQueryRequest(BaseModel):
    """Request body for POST /sessions/{session_id}/query/sql."""

    model_id: str
    query: QueryObject
    dialect: str | None = Field(
        default=None,
        description=(
            "SQL dialect. Resolution: explicit value → model.settings.defaultDialect → "
            "DB_VENDOR env → 'postgres'."
        ),
    )

QueryCompileResponse

Bases: BaseModel

Response body for POST /query/sql.

Source code in src/orionbelt/api/schemas.py
class QueryCompileResponse(BaseModel):
    """Response body for POST /query/sql."""

    sql: str
    dialect: str
    resolved: ResolvedInfoResponse
    warnings: list[StructuredWarning] = Field(default_factory=list)
    sql_valid: bool = True
    explain: ExplainPlanResponse | None = None
    physical_tables: list[str] = Field(
        default_factory=list,
        description=(
            "Deduplicated DATABASE.SCHEMA.CODE strings the query touches. "
            "Drives freshness-cache TTL composition and heartbeat invalidation."
        ),
    )

ValidateRequest

Bases: BaseModel

Request body for POST /validate.

Source code in src/orionbelt/api/schemas.py
class ValidateRequest(BaseModel):
    """Request body for POST /validate."""

    model_yaml: str | None = Field(
        default=None,
        description="OBML model as YAML string (provide model_yaml OR model_json)",
        max_length=5_000_000,
    )
    model_json: dict[str, object] | str | None = Field(
        default=None,
        description="OBML model as JSON object or JSON string (auto-parsed)",
    )
    extends: list[str] | None = Field(
        default=None,
        description="Optional inline YAML strings of analytical fragments to merge",
    )
    inherits: str | None = Field(
        default=None,
        description="Optional model ID of an already-loaded parent model in the session",
    )

    @model_validator(mode="after")
    def _parse_model_json_string(self) -> ValidateRequest:
        if isinstance(self.model_json, str):
            self.model_json = json.loads(self.model_json)
        return self

ValidateResponse

Bases: BaseModel

Response body for POST /validate.

Source code in src/orionbelt/api/schemas.py
class ValidateResponse(BaseModel):
    """Response body for POST /validate."""

    valid: bool
    errors: list[ErrorDetail] = Field(default_factory=list)
    warnings: list[ErrorDetail] = Field(default_factory=list)

DialectListResponse

Bases: BaseModel

Response for GET /dialects.

Source code in src/orionbelt/api/schemas.py
class DialectListResponse(BaseModel):
    """Response for GET /dialects."""

    dialects: list[DialectInfo] = Field(default_factory=list)

HealthResponse

Bases: BaseModel

Health check response.

Source code in src/orionbelt/api/schemas.py
class HealthResponse(BaseModel):
    """Health check response."""

    status: str = "ok"
    version: str = ""
    auth_mode: str = Field(
        default="none",
        description="Effective auth mode: 'none', 'api_key', or 'oidc'. "
        "Clients check this to know whether a credential is required.",
    )

Settings

orionbelt.settings.Settings

Bases: BaseSettings

Configuration for OrionBelt REST API server.

Values are read from environment variables and from a .env file in the working directory. See .env.template for all options.

Source code in src/orionbelt/settings.py
class Settings(BaseSettings):
    """Configuration for OrionBelt REST API server.

    Values are read from environment variables and from a ``.env`` file
    in the working directory.  See ``.env.template`` for all options.
    """

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    # Shared
    log_level: str = "INFO"
    # Log format:
    #   "console"  — pretty-printed for local dev (default)
    #   "json"     — structured JSON for log aggregators (ELK, Datadog, etc.)
    #   "cloudrun" — JSON + disables uvicorn access logs (Cloud Run provides its own)
    log_format: str = "console"

    # REST API
    api_server_host: str = "localhost"
    api_server_port: int = 8000
    port: int | None = None  # Cloud Run injects PORT; takes precedence over api_server_port

    # Authentication. Single AUTH_MODE selector drives every direct surface
    # (REST now; Flight + pgwire in Phase 2). Off by default to preserve the
    # public-demo / local-dev behaviour. See design/PLAN_authentication.md §1.
    #   "none"    — no auth (default)
    #   "api_key" — validate API_KEYS against the shared key store
    #   "oidc"    — Phase 4 (not implemented; rejected loudly at startup)
    auth_mode: str = "none"
    api_keys: str = ""  # comma-separated; required when auth_mode=api_key (>=16 chars each)
    api_key_header: str = "X-API-Key"  # REST header name; Bearer is always accepted as fallback
    # Legacy alias for auth_mode=api_key. Deprecated; honoured one release with
    # a startup warning. Only takes effect when AUTH_MODE is left at "none".
    auth_enabled: bool = False

    # Public-doc surfaces. Default True preserves current public-demo behaviour.
    # Set EXPOSE_API_DOCS=false on non-demo deployments to disable Swagger UI,
    # ReDoc, and the OpenAPI schema endpoint. EXPOSE_OPENAPI_SCHEMA can be
    # toggled independently to keep /openapi.json live (e.g. for client codegen)
    # while hiding the human-facing /docs and /redoc pages.
    expose_api_docs: bool = True
    expose_openapi_schema: bool = True

    @property
    def effective_port(self) -> int:
        """Return the port to listen on (Cloud Run PORT takes precedence)."""
        return self.port if self.port is not None else self.api_server_port

    # Sessions
    session_ttl_seconds: int = 1800  # 30 min inactivity
    session_max_age_seconds: int = 86400  # 24 h absolute max lifetime
    session_cleanup_interval: int = 60  # seconds between cleanup sweeps
    max_sessions: int = 500  # global concurrent session cap (429 when full)
    max_models_per_session: int = 10  # max models a single session may hold
    disable_session_list: bool = False  # hide GET /sessions endpoint
    session_rate_limit: int = 10  # max POST /sessions per IP per minute
    trusted_proxy_count: int = 0  # number of trusted reverse proxies in front of the app

    # Admin-curated model pre-loading. When MODEL_FILES is set, REST POST
    # /models returns 403 (the catalog is admin-managed) and the models are
    # loaded into named protected sessions at startup.
    #
    # MODEL_FILES (comma-separated paths):
    #     Each OBML YAML loads into its own internal session, addressable
    #     by the OBML `name:` field (fallback: filename stem, normalized to
    #     a valid identifier). BI tools select via the Flight `database`
    #     catalog or pgwire `database=` URL parameter. A single path is
    #     fine — it just means one named protected session.
    #     See design/PLAN_flight_natural_sql.md §3.x multi-model.
    model_dir: str | None = None  # base directory (set by Docker)
    model_files: str | None = None  # comma-separated paths

    # Query execution
    query_execute: bool = False  # enable POST /v1/query/execute
    query_default_limit: int = 1000  # max rows when query has no LIMIT
    db_pool_size: int = 5  # connection pool size per dialect

    # Default locale for /v1/query/execute?format_values=true (and TSV output).
    # Used when the request omits the ``locale`` query param. BCP-47 tag
    # (e.g. "de", "en-US"). Empty → en-style separators ("," / ".").
    default_locale: str = ""

    # Arrow Flight SQL server (requires ob-flight-extension)
    flight_enabled: bool = False  # start gRPC Flight server on FLIGHT_PORT (implies query_execute)
    flight_port: int = 8815
    flight_auth_mode: str = "none"  # "none" or "token"
    flight_api_token: str | None = None
    # TLS for the Flight listener. Paths, not PEM content: a private key in the
    # environment is readable by ``docker inspect`` and by a crash dump.
    # Both or neither - a cert without a key is a configuration error rather
    # than a quiet fall back to plaintext, which would leave a deployment
    # believing it has TLS.
    flight_tls_cert: str | None = None
    flight_tls_key: str | None = None
    # Mutual TLS: the CA that client certificates are validated against.
    # Setting it requires client certs.
    flight_tls_client_ca: str | None = None
    db_vendor: str = "duckdb"  # default vendor driver for Flight query execution

    # Flight Semantic QL governance. See design/PLAN_flight_natural_sql.md.
    # Semantic QL / OBSQL (SELECT dim, measure FROM <model>) is always enabled.
    # Raw SQL pass-through and write operations are **not** configurable —
    # OBSL is a semantic layer, not a JDBC proxy. There are no env flags
    # that allow arbitrary SQL through to the warehouse.

    # Postgres wire surface. Today: trust auth only, simple-query protocol.
    # Auth modes "password" / "scram-sha-256" land in Phase 2 alongside the
    # shared auth subsystem (see design/PLAN_authentication.md §3.3).
    pgwire_enabled: bool = False
    pgwire_host: str = "0.0.0.0"  # noqa: S104 — server bind address
    pgwire_port: int = 5432
    pgwire_auth_mode: str = "trust"  # "trust" (Step 1) | "password" | "scram-sha-256" (Step 6)
    # TLS for the pgwire listener, same shape as the Flight settings and read
    # by the same loader. Postgres negotiates with an SSLRequest rather than
    # starting encrypted, so with these set the server answers ``S`` and
    # upgrades the socket; without them it answers ``N`` as before.
    # TLS for the REST listener, same shape and same loader as the two wire
    # surfaces. Unlike those, REST is usually behind something that terminates
    # TLS already (Cloud Run, an ingress, nginx) - these are for when it is not:
    # a LAN or on-premise deployment with no proxy, or a requirement that the
    # internal hop be encrypted too rather than only the edge.
    api_tls_cert: str | None = None
    api_tls_key: str | None = None
    api_tls_client_ca: str | None = None
    pgwire_tls_cert: str | None = None
    pgwire_tls_key: str | None = None
    pgwire_tls_client_ca: str | None = None
    pgwire_max_connections: int = 64
    pgwire_query_timeout_seconds: int = 60
    # Hard deadline for the pre-auth handshake (startup + password/SCRAM
    # exchange). Bounds how long an unauthenticated client can hold a
    # connection slot, preventing slot-exhaustion DoS.
    pgwire_auth_timeout_seconds: int = 10

    # One-shot batch endpoint (POST /v1/oneshot/batch). See PLAN_oneshot_batch.md.
    oneshot_batch_max_queries: int = 50
    oneshot_batch_max_parallelism: int = 8
    oneshot_batch_default_timeout_ms: int = 30000  # per-query
    oneshot_batch_batch_timeout_ms: int = 120000  # whole batch

    # Freshness-driven result cache. See design/PLAN_freshness_driven_cache.md.
    cache_backend: str = "noop"  # "noop" or "file"
    cache_dir: str = "./cache"
    cache_max_ttl_seconds: int = 86400
    cache_min_ttl_seconds: int = 5
    cache_max_value_bytes: int = 10 * 1024 * 1024  # 10 MB
    cache_max_disk_bytes: int = 5 * 1024 * 1024 * 1024  # 5 GB
    cache_sweep_interval_seconds: int = 86400
    cache_unknown_freshness_policy: str = "no_cache"  # or "default_ttl"
    cache_unknown_freshness_default_ttl: int = 300
    heartbeat_auth_token: str | None = None  # endpoint disabled (404) when unset

effective_port property

Return the port to listen on (Cloud Run PORT takes precedence).