qonto-assistant/src/qonto_assistant/config.py

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path


def _repo_root() -> Path:
    return Path(__file__).resolve().parents[2]


@dataclass(frozen=True, slots=True)
class Settings:
    service_name: str
    default_tenant_id: str
    default_actor_lane: str
    required_scope: str
    enforce_scope: bool
    policy_file: Path
    qonto_base_url: str
    qonto_fixture_dir: Path | None
    qonto_auth_mode: str
    qonto_organization_path: str
    qonto_transactions_path: str
    qonto_timeout_seconds: float
    qonto_max_retries: int
    qonto_secret_ttl_seconds: int
    rate_limit_requests: int
    rate_limit_window_seconds: int
    max_concurrency: int
    deny_escalation_enabled: bool
    deny_escalation_threshold: int
    deny_escalation_window_seconds: int
    deny_escalation_lockout_seconds: int
    key_cape_jwks_url: str | None
    key_cape_issuer: str
    key_cape_audience: str
    key_cape_required: bool
    key_cape_cache_seconds: float
    key_cape_timeout_seconds: float
    flex_auth_base_url: str | None
    flex_auth_timeout_seconds: float
    tenant_engine_base_url: str | None
    tenant_engine_timeout_seconds: float
    tenant_engine_required_roles: frozenset[str]
    credential_source: str
    openbao_path: str
    openbao_command: str
    openbao_timeout_seconds: float
    mcp_auth_token: str | None
    host: str
    port: int

    @classmethod
    def from_env(cls) -> "Settings":
        policy_file = Path(
            os.getenv(
                "QONTO_ASSISTANT_POLICY_FILE",
                str(_repo_root() / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"),
            )
        )
        return cls(
            service_name=os.getenv("QONTO_ASSISTANT_SERVICE_NAME", "qonto-assistant"),
            default_tenant_id=os.getenv("QONTO_ASSISTANT_DEFAULT_TENANT", "binky"),
            default_actor_lane=os.getenv("QONTO_ASSISTANT_DEFAULT_LANE", "green"),
            required_scope=os.getenv("QONTO_ASSISTANT_REQUIRED_SCOPE", "finance.qonto.read"),
            enforce_scope=os.getenv("QONTO_ASSISTANT_ENFORCE_SCOPE", "false").lower() == "true",
            policy_file=policy_file,
            qonto_base_url=os.getenv("QONTO_BASE_URL", "https://thirdparty.qonto.com"),
            qonto_fixture_dir=(
                Path(os.environ["QONTO_FIXTURE_DIR"]).resolve()
                if os.getenv("QONTO_FIXTURE_DIR")
                else None
            ),
            qonto_auth_mode=os.getenv("QONTO_AUTH_MODE", "legacy_api_key"),
            qonto_organization_path=os.getenv("QONTO_ORGANIZATION_PATH", "/v2/organization"),
            qonto_transactions_path=os.getenv("QONTO_TRANSACTIONS_PATH", "/v2/transactions"),
            qonto_timeout_seconds=float(os.getenv("QONTO_TIMEOUT_SECONDS", "10")),
            qonto_max_retries=int(os.getenv("QONTO_MAX_RETRIES", "1")),
            qonto_secret_ttl_seconds=int(os.getenv("QONTO_SECRET_TTL_SECONDS", "300")),
            rate_limit_requests=int(os.getenv("QONTO_RATE_LIMIT_REQUESTS", "20")),
            rate_limit_window_seconds=int(os.getenv("QONTO_RATE_LIMIT_WINDOW_SECONDS", "60")),
            max_concurrency=int(os.getenv("QONTO_MAX_CONCURRENCY", "4")),
            deny_escalation_enabled=os.getenv("QONTO_DENY_ESCALATION_ENABLED", "true").lower() == "true",
            deny_escalation_threshold=int(os.getenv("QONTO_DENY_ESCALATION_THRESHOLD", "3")),
            deny_escalation_window_seconds=int(os.getenv("QONTO_DENY_ESCALATION_WINDOW_SECONDS", "60")),
            deny_escalation_lockout_seconds=int(os.getenv("QONTO_DENY_ESCALATION_LOCKOUT_SECONDS", "300")),
            key_cape_jwks_url=os.getenv("QONTO_KEY_CAPE_JWKS_URL") or None,
            key_cape_issuer=os.getenv("QONTO_KEY_CAPE_ISSUER", "https://key-cape.netkingdom"),
            key_cape_audience=os.getenv("QONTO_KEY_CAPE_AUDIENCE", "qonto-assistant"),
            key_cape_required=os.getenv("QONTO_KEY_CAPE_REQUIRED", "false").lower() == "true",
            key_cape_cache_seconds=float(os.getenv("QONTO_KEY_CAPE_CACHE_SECONDS", "300")),
            key_cape_timeout_seconds=float(os.getenv("QONTO_KEY_CAPE_TIMEOUT_SECONDS", "5")),
            flex_auth_base_url=os.getenv("QONTO_FLEX_AUTH_URL") or None,
            flex_auth_timeout_seconds=float(os.getenv("QONTO_FLEX_AUTH_TIMEOUT_SECONDS", "3")),
            tenant_engine_base_url=os.getenv("QONTO_TENANT_ENGINE_URL") or None,
            tenant_engine_timeout_seconds=float(os.getenv("QONTO_TENANT_ENGINE_TIMEOUT_SECONDS", "3")),
            tenant_engine_required_roles=frozenset(
                role.strip()
                for role in os.getenv("QONTO_TENANT_ENGINE_REQUIRED_ROLES", "VEN,CUS").split(",")
                if role.strip()
            ),
            credential_source=os.getenv("QONTO_CREDENTIAL_SOURCE", "env"),
            openbao_path=os.getenv("QONTO_OPENBAO_PATH", "tenants/binky/qonto-api"),
            openbao_command=os.getenv("QONTO_OPENBAO_COMMAND", "bao"),
            openbao_timeout_seconds=float(os.getenv("QONTO_OPENBAO_TIMEOUT_SECONDS", "5")),
            mcp_auth_token=os.getenv("QONTO_ASSISTANT_MCP_TOKEN") or None,
            host=os.getenv("QONTO_ASSISTANT_HOST", "127.0.0.1"),
            port=int(os.getenv("QONTO_ASSISTANT_PORT", "8080")),
        )