Source code for manna.config

from functools import lru_cache
from typing import Literal

from pydantic import Field, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict




class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="MANNA_",
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    host: str = "0.0.0.0"
    port: int = 8000
    # Optional comma-separated allow-list of archive short_names
    # (e.g. "datalab,alma"). Unset/empty => every archive physically present in
    # the `archives/` package is active. See archives/__init__.py and
    # docs/archives-spec.md.
    archives: str | None = None
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
    # Optional comma-separated allow-list of hostnames the server may fetch
    # (e.g. "almascience.eso.org,eso.org"). Matches exactly or as a subdomain.
    # Unset => any *public* host is reachable, which keeps search_ivoa_registry
    # discovery working; private/loopback/link-local space is refused either
    # way. See _url_guard.py.
    allowed_hosts: str | None = None
    # Slice 5: async TAP family.
    tap_sync_timeout_seconds: float = 20.0
    # count_observations_near_target: bounded budget for polling an async count job
    # (an archive whose count_target runs async). A COUNT returns one row and
    # completes fast once scheduled; on budget exhaustion the tool returns a
    # pending envelope with the job_url.
    count_async_budget_seconds: float = Field(default=15.0, gt=0)
    count_async_poll_interval_seconds: float = Field(default=1.0, gt=0)
    # Bounded server-side wait on async TAP jobs (tools/tap.py::_wait_for_phase).
    # get_async_job_status blocks up to async_wait_seconds when the call omits
    # wait_seconds; a caller-supplied value is clamped to async_wait_max_seconds.
    # The run_adql_query async promotion waits the default before returning its
    # envelope. Keep the max under the MCP SDK's 60 s client request timeout —
    # the worst single call is a 20 s sync attempt plus the promotion wait.
    async_wait_seconds: float = Field(default=20.0, ge=0)
    async_wait_max_seconds: float = Field(default=30.0, ge=0)
    # Inline response caps (results.py). A TAP result larger than EITHER limit
    # is routed to an async job whose result the client fetches itself (the
    # server never holds the bytes); discovery tools (cone / SIA search)
    # truncate inline instead. Defaults are sized for small-context backends
    # (e.g. a 131072-token local vLLM), where a single fat inline result can
    # overflow the model window. Raise them for frontier models with large
    # context windows.
    inline_row_limit: int = 200
    inline_byte_limit: int = 48 * 1024
    # describe_ivoa_service degrades from full per-column detail to a table
    # catalog (names + descriptions + column counts) once the full introspection
    # payload would exceed this many bytes. Prevents a large service (e.g. Gaia,
    # ~127k tokens of tables × columns) from overflowing the model context. See
    # shape_registry_describe_result.
    registry_describe_byte_limit: int = 48 * 1024

    @model_validator(mode="after")
    def _async_wait_default_within_max(self) -> "Settings":
        if self.async_wait_seconds > self.async_wait_max_seconds:
            raise ValueError(
                "MANNA_ASYNC_WAIT_SECONDS must not exceed MANNA_ASYNC_WAIT_MAX_SECONDS"
            )
        return self





@lru_cache(maxsize=1)
def get_settings() -> Settings:
    """Process-wide Settings singleton.

    Cached so runtime consumers (the lazy backend accessors, the URL
    guard) read environment / .env once rather than re-parsing per call.
    Tests that mutate the environment must call ``get_settings.cache_clear()``
    to force a re-read.
    """
    return Settings()