mirror of
https://github.com/migatu/astrololo.git
synced 2026-07-14 21:38:37 +00:00
gitea
16d35c16dc
Trzy niezależne usługi FastAPI komunikujące się przez HTTP/JSON, każda zna tylko adres warstwy bezpośrednio pod nią: - presentation (:8000) — strona WWW + formularz - logic (:8001) — reguły biznesowe, pośrednik - data (:8002) — wyszukiwanie danych za interfejsem DataProvider Warstwa danych: czytanie setek plików .xlsx z wykrywaniem nagłówka i mapowaniem układu kolumn na schemat kanoniczny, z 4-poziomowym cache (schemat L1, Parquet L2, wyniki zapytań L3, odwrócony indeks L4) i unieważnianiem po odcisku pliku. Gotowa ścieżka migracji do SQL (ingest/to_sql.py + SqlDataProvider, przełączane przez DATA_PROVIDER). Zawiera docker-compose, Makefile, generator danych przykładowych. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
"""SqlDataProvider — implementacja docelowa (po migracji z Excela).
|
|
|
|
Szkielet. Realizuje TEN SAM interfejs DataProvider, więc przełączenie to tylko
|
|
zmiana zmiennej środowiskowej DATA_PROVIDER=sql (patrz factory.py). Warstwa
|
|
logiczna i prezentacji nie zmieniają ani jednej linii.
|
|
|
|
Dane ładuje do bazy skrypt ingest/to_sql.py (ten sam loader Excela -> tabele SQL).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from sqlalchemy import create_engine, text
|
|
|
|
from app.config import Settings
|
|
from app.models import HealthInfo, SearchQuery, SearchResult
|
|
from app.providers.base import DataProvider
|
|
|
|
|
|
class SqlDataProvider(DataProvider):
|
|
name = "sql"
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.s = settings
|
|
self.engine = create_engine(settings.sql_url, future=True)
|
|
|
|
def search(self, query: SearchQuery) -> SearchResult:
|
|
t0 = time.perf_counter()
|
|
op = "=" if query.exact else "LIKE"
|
|
val = query.value if query.exact else f"%{query.value}%"
|
|
cols = ", ".join(query.fields) if query.fields else "*"
|
|
# UWAGA: nazwy kolumn/tabel walidować względem białej listy schematu.
|
|
sql = text(f"SELECT {cols} FROM records WHERE {query.key} {op} :v LIMIT :lim")
|
|
with self.engine.connect() as conn:
|
|
rows = [dict(r._mapping) for r in conn.execute(sql, {"v": val, "lim": query.limit})]
|
|
return SearchResult(
|
|
rows=rows,
|
|
total=len(rows),
|
|
elapsed_ms=round((time.perf_counter() - t0) * 1000, 2),
|
|
cache="miss",
|
|
provider=self.name,
|
|
)
|
|
|
|
def health(self) -> HealthInfo:
|
|
with self.engine.connect() as conn:
|
|
n = conn.execute(text("SELECT COUNT(*) FROM records")).scalar() or 0
|
|
return HealthInfo(provider=self.name, indexed_files=0, details={"records": int(n)})
|