astrololo/services/logic/app/clients/data_client.py

"""Klient HTTP do warstwy bazodanowej.

Jedyny punkt styku w dół. Gdyby warstwa bazodanowa zmieniła implementację
(Excel→SQL), tutaj nie zmienia się NIC — kontrakt /search jest stały.
"""
from __future__ import annotations

import os
from typing import Any

import httpx

from app import link_crypto
from app.config import settings


def _auth_headers() -> dict[str, str]:
    """Token międzywarstwowy (LOG-32) — pusty, gdy ochrona wyłączona."""
    token = os.getenv("INTERNAL_TOKEN", "")
    return {"X-Astrololo-Token": token} if token else {}


def _link() -> link_crypto.Link | None:
    """Klucz łącza logika↔dane. Czytany przy każdym wywołaniu, bo konfiguracja
    może się zmienić bez restartu procesu (testy, podmiana sekretu)."""
    key = link_crypto.key_from_env(link_crypto.ENV_LOGIC_DATA)
    return link_crypto.Link(key) if key else None


class DataClient:
    def __init__(self, base_url: str | None = None) -> None:
        self.base_url = (base_url or settings.data_url).rstrip("/")

    def search(
        self,
        key: str,
        value: str,
        exact: bool,
        limit: int,
        fields: list[str] | None = None,
    ) -> dict[str, Any]:
        payload = {"key": key, "value": value, "exact": exact, "limit": limit, "fields": fields}
        with httpx.Client(timeout=max(settings.http_timeout, 30.0)) as client:
            return link_crypto.call_json(client, "POST", f"{self.base_url}/search",
                                         payload=payload, headers=_auth_headers(),
                                         link=_link())

    def health(self) -> dict[str, Any]:
        # /health celowo poza szyfrowaniem — pukają tu sondy k8s, które klucza
        # nie mają, a nie przechodzi tędy nic z baz.
        with httpx.Client(timeout=settings.http_timeout) as client:
            r = client.get(f"{self.base_url}/health", headers=_auth_headers())
            r.raise_for_status()
            return r.json()