gitea
4c1e7f8808
Testy / Testy warstwy logicznej (silnik) (pull_request) Successful in 10m30s
Testy / Testy warstwy prezentacji (dostęp do baz) (pull_request) Successful in 9m29s
Testy / Testy warstwy bazodanowej (ochrona baz) (pull_request) Successful in 9m28s
Testy / Build obrazu silnika B (swisseph) (pull_request) Successful in 12s
Testy / Kontrola składni wszystkich warstw (pull_request) Successful in 8s
build / build (push) Successful in 40s
Testy / Testy warstwy logicznej (silnik) (push) Successful in 11m3s
Testy / Testy warstwy prezentacji (dostęp do baz) (push) Successful in 9m35s
Testy / Testy warstwy bazodanowej (ochrona baz) (push) Successful in 9m29s
Testy / Build obrazu silnika B (swisseph) (push) Successful in 14s
Testy / Kontrola składni wszystkich warstw (push) Successful in 10s
Wymaganie przedefiniowane pod model serwerowy (#47): nie wybiera się folderu — pliki leżą na stałym NFS. Potrzeba za to WIDZIEĆ, jakie bazy są dostępne i móc zdecydować, które biorą udział w interpretacji. Warstwa danych: `bases.py` (lista plików + metaopis: nazwa, ścieżka, rozmiar, data, stan) i endpoint `/bases`. Wyłączone bazy są ODSIEWANE z kandydatów przy wyszukiwaniu, więc naprawdę nie biorą udziału w interpretacji — nie tylko znikają z listy. Lista wyłączonych wchodzi do klucza cache zapytań: bez tego zmiana ustawień oddawałaby wynik sprzed zmiany, czyli treść bazy uznanej za wyłączoną. `list_bases()` doszło do interfejsu dostawcy jako OPCJONALNE (SQL nie operuje na plikach → pusto, zamiast wywrotki). Przelot logika → prezentacja i ekran „Ustawienia" z tabelą baz. Przez łącze idą SAME METADANE — podgląd listy nie jest kolejną drogą do wyniesienia treści. Stan przełączników jest DEKLARATYWNY (`DISABLED_BASES`), nie klikalny — i to jest świadome: udział z bazami montujemy read-only, a katalog cache to `emptyDir`, więc zapisany przełącznik ginąłby przy restarcie poda i po cichu włączał z powrotem wyłączoną bazę. Ekran mówi wprost, jak wyłączyć bazę i dlaczego nie klikaniem. Tryb klikalny wymagałby dołożenia trwałego wolumenu. Weryfikacja na żywym łańcuchu: `/bases` przechodzi przez SZYFROWANE łącze (logic→data), pokazuje 3 bazy z metaopisem i stanem; wyszukiwanie daje 3 → 2 → 0 wierszy w miarę wyłączania baz. Testy: dane +6, prezentacja +6. Dane 13, logika 277, prezentacja 249. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
"""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 bases(self) -> dict[str, Any]:
|
|
"""Lista baz na udziale + metaopis (DAN-15) — same metadane, bez treści."""
|
|
with httpx.Client(timeout=settings.http_timeout) as client:
|
|
return link_crypto.call_json(client, "GET", f"{self.base_url}/bases",
|
|
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()
|