SPB Git
spboucher.ai

spb/ai-atlas

Public

API v1: FastAPI routers per docs/API.md (public + admin), detail builder, hardware fit, entity merge, indexes, tests

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed (Sep 11, 2026) parent e6b6b4a Browse files

29 changed files +3,317 −4

29 changed files
Load all 29 files Unified Split
modified docs/API.md +31 −0
@@ -117,3 +117,34 @@ EntitySummary & {
117 | `GET /admin/infrastructure` | heartbeats, archive size, DB size (`pg_database_size`), table sizes, node hostname, uptime | 117 | `GET /admin/infrastructure` | heartbeats, archive size, DB size (`pg_database_size`), table sizes, node hostname, uptime |
118 | `POST /admin/cache/flush` | `{ flushed: n }` | 118 | `POST /admin/cache/flush` | `{ flushed: n }` |
119 | `POST /admin/stats/recompute` · `POST /admin/quality/recompute` | maintenance | 119 | `POST /admin/stats/recompute` · `POST /admin/quality/recompute` | maintenance |
120+
121+## Implementation notes (2026-09-11, `src/aiatlas/api`)
122+
123+Deviations and additions relative to the table above — everything else is implemented as written.
124+
125+- **Validation errors**: parameters rejected by FastAPI/pydantic typing or bounds (`limit=999`, non-integer `depth`…) return **422** with
126+ `{ detail: "<field>: <message>", errors: [...] }`; semantic errors raised by the routes (bad `sort`, bad date, wrong `scope`, `ids` count) return **400**.
127+- **`/companies`** (listing, alias, `/compare` type `company`) covers the organization-like types `company | organization | lab | university`.
128+ `/frameworks/{slug}` also accepts `library | runtime`; `/tools/{slug}` accepts `tool | agent | application | product | mcp_server`.
129+- **`/explore/{type}`** accepts plural forms (`/explore/models`, `/explore/companies`) and adds `entity_type` + `label` to the page.
130+- **`EntitySummary.description`** is truncated at 280 characters in summaries (full text in `EntityDetail`).
131+- **Cursor feeds** (`/changes`, `/entities/{slug}/timeline`) add `next_before` (the `observed_at` of the last item, or `null` at the end).
132+ `/changes` also honours `offset` and `entity=` for convenience; `total` is capped at 10 000.
133+- **`/changes/daily`** adds `total`, `labels`, `previous_day`, `next_day`; sections are capped at `per_section` (default 30) items each.
134+- **`/timeline`** month groups carry `count`; the response includes `total` (events returned).
135+- **`/search`**: `query` carries `semantic: bool` (embedding used or not). When the embedding endpoint times out (2 s) the API backs off to
136+ FTS-only for 5 minutes (`aia:api:search:embed-degraded`) so the LLM never adds latency to consecutive queries.
137+- **`hardware_fit`** (model detail) is *omitted* when `parameter_count` is unknown (nothing is estimated from thin air); when present the
138+ detail also carries `hardware_fit_assumptions: string[]`. `/hardware/fit` adds `estimated: true`, `counts`, optional `openness=` filter.
139+- **`/prices/index`** series rows add `max_input` and `offers`; response adds `days` and `note`. `/prices/history` requires `model` or `provider`.
140+- **`/benchmarks/{slug}/results`** and `/history` include the `benchmark` summary; result rows include `valid_to`.
141+- **`/diff`** adds `scope` (resolved description) and extra `counts` (`events`, `price_rows_opened/closed`, `claims_superseded`, `entities_at_a/b`).
142+- **`/sources`** items add `snapshots`, `claims`, `base_url`, `robots_policy`, `priority`, `notes`; response adds `tiers` legend.
143+- **`/methodology`** adds `quality_version`, `expected_fields`, `principles`; `event_types` are live counts from `change_events`.
144+- **`/trending`** accepts `type=`; `/sitemap` `limit` ≤ 5000 (the one listing not capped at 200).
145+- **Rate limits** are in-process sliding windows (search 60/min, views 1/s, admin 240/min per IP) — fine for the single uvicorn worker behind Next.js.
146+- **Admin**: `GET /admin/review?status=all` lists every status; approving a `conflict` with `resolution.keep_claim_id` promotes that claim;
147+ `PATCH /admin/connectors/{name}` also resets `health`/`circuit_open_until` when toggling; `/admin/connectors` flags `run_now_pending` and
148+ connectors present in code but not in the table; `/admin/snapshots/{id}` never returns `raw_path`/`text_path` (`has_raw`/`has_text` booleans instead).
149+- **Migration `0002_api_indexes`** adds partial indexes for the read paths (type + release_date/family/openness/updated/first_seen, importance-ordered
150+ events, current leaderboards/prices, page views, review kinds).
added migrations/versions/0002_api_indexes.py +63 −0
@@ -0,0 +1,63 @@
1+"""API read-path indexes: attribute-driven listings (release_date, family, openness), importance-ordered change feeds,
2+current benchmark leaderboards, page-view trending, review queue by kind.
3+
4+Revision ID: 0002
5+"""
6+from __future__ import annotations
7+
8+from alembic import op
9+
10+revision = "0002"
11+down_revision = "0001"
12+branch_labels = None
13+depends_on = None
14+
15+SQL = r"""
16+create index if not exists entities_type_release_idx on entities (entity_type, (attributes->>'release_date') desc nulls last) where merged_into is null;
17+create index if not exists entities_type_family_idx on entities (entity_type, (attributes->>'family')) where merged_into is null;
18+create index if not exists entities_type_openness_idx on entities (entity_type, (attributes->>'openness')) where merged_into is null;
19+create index if not exists entities_type_updated_idx on entities (entity_type, updated_at desc) where merged_into is null;
20+create index if not exists entities_type_first_seen_idx on entities (entity_type, first_seen_at desc) where merged_into is null;
21+create index if not exists entities_slug_id_idx on entities (id) include (slug);
22+create index if not exists change_events_importance_idx on change_events (importance desc, observed_at desc);
23+create index if not exists change_events_type_observed_idx on change_events (observed_at desc) where event_type <> 'DOCUMENT_CHANGED';
24+create index if not exists benchmark_results_current_idx on benchmark_results (benchmark_id, higher_is_better, score) where valid_to is null;
25+create index if not exists benchmark_results_model_current_idx on benchmark_results (model_id) where valid_to is null;
26+create index if not exists prices_current_model_idx on prices (model_id, provider_id) where valid_to is null;
27+create index if not exists prices_valid_range_idx on prices (valid_from, valid_to);
28+create index if not exists page_views_day_idx on page_views (day desc, views desc);
29+create index if not exists review_queue_kind_idx on review_queue (kind, status, created_at desc);
30+create index if not exists documents_source_idx on documents (source_id, last_fetched_at desc);
31+create index if not exists claims_entity_valid_idx on claims (entity_id, valid_from, valid_to);
32+create index if not exists relations_predicate_idx on relations (predicate, subject_id, object_id) where valid_to is null;
33+"""
34+
35+
36+def upgrade() -> None:
37+ for statement in _split(SQL):
38+ op.execute(statement)
39+
40+
41+def downgrade() -> None:
42+ raise RuntimeError("AI Atlas migrations are forward-only: historical data is never disposable")
43+
44+
45+def _split(sql: str) -> list[str]:
46+ out: list[str] = []
47+ buf: list[str] = []
48+ in_dollar = False
49+ for line in sql.splitlines():
50+ stripped = line.strip()
51+ if stripped.count("$$") % 2 == 1:
52+ in_dollar = not in_dollar
53+ buf.append(line)
54+ if not in_dollar and stripped.endswith(";"):
55+ body = [ln for ln in buf if not ln.strip().startswith("--") or in_dollar]
56+ stmt = "\n".join(body).strip()
57+ if stmt:
58+ out.append(stmt)
59+ buf = []
60+ tail = "\n".join(ln for ln in buf if not ln.strip().startswith("--")).strip()
61+ if tail:
62+ out.append(tail)
63+ return out
added src/aiatlas/api/__init__.py +1 −0
@@ -0,0 +1 @@
1+"""AI Atlas public + admin API (FastAPI, `/api/v1`). Contract: docs/API.md."""
added src/aiatlas/api/common.py +412 −0
@@ -0,0 +1,412 @@
1+"""Shared API helpers: errors, pagination, date parsing, Redis response cache, admin auth, IP rate limits, serialisers.
2+
3+Every route returns plain dicts; `AtlasJSONResponse` renders them with orjson (datetimes → ISO-8601 UTC, Decimal → float)."""
4+from __future__ import annotations
5+
6+import hmac
7+import inspect
8+import time
9+from collections import defaultdict, deque
10+from collections.abc import Awaitable, Callable
11+from datetime import UTC, date, datetime
12+from datetime import time as dtime
13+from decimal import Decimal
14+from functools import wraps
15+from typing import Any
16+
17+import orjson
18+from fastapi import Depends, HTTPException, Query, Request
19+from fastapi.responses import JSONResponse
20+from sqlalchemy.ext.asyncio import AsyncConnection
21+
22+from aiatlas.config import settings
23+from aiatlas.db import fetch_one
24+from aiatlas.services import cache
25+
26+# ------------------------------------------------------------------------------------------------------------------ errors & JSON
27+
28+
29+class ApiError(HTTPException):
30+ """Always `{"detail": "..."}` — never a stack trace."""
31+
32+ def __init__(self, status: int, detail: str):
33+ super().__init__(status_code=status, detail=detail)
34+
35+
36+def _json_default(obj: Any) -> Any:
37+ if isinstance(obj, Decimal):
38+ return float(obj)
39+ if isinstance(obj, (set, frozenset)):
40+ return sorted(obj, key=str)
41+ if isinstance(obj, bytes):
42+ return obj.decode("utf-8", "replace")
43+ return str(obj)
44+
45+
46+_ORJSON_OPTS = orjson.OPT_NON_STR_KEYS | orjson.OPT_UTC_Z | orjson.OPT_NAIVE_UTC
47+
48+
49+def dumps(value: Any) -> bytes:
50+ return orjson.dumps(value, default=_json_default, option=_ORJSON_OPTS)
51+
52+
53+def normalize(value: Any) -> Any:
54+ """Round-trip through orjson so cached and fresh responses are byte-identical (datetimes → strings, Decimal → float)."""
55+ return orjson.loads(dumps(value))
56+
57+
58+class AtlasJSONResponse(JSONResponse):
59+ media_type = "application/json"
60+
61+ def render(self, content: Any) -> bytes:
62+ return dumps(content)
63+
64+
65+# ------------------------------------------------------------------------------------------------------------------ pagination & parsing
66+
67+
68+
69+class Pagination:
70+ def __init__(self, limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0)):
71+ self.limit = limit
72+ self.offset = offset
73+
74+
75+PAGINATION = Depends() # `p: Pagination = PAGINATION` — FastAPI resolves the class from the annotation
76+
77+
78+def page(items: list[Any], total: int, p: Pagination) -> dict[str, Any]:
79+ return {"items": items, "total": int(total), "limit": p.limit, "offset": p.offset}
80+
81+
82+def parse_date(v: str | None, name: str = "date") -> date | None:
83+ """YYYY-MM-DD (or ISO datetime) → `date`. asyncpg needs real date objects — never cast text in SQL."""
84+ if v is None or v == "":
85+ return None
86+ s = v.strip().replace("Z", "+00:00")
87+ try:
88+ if len(s) == 10:
89+ return date.fromisoformat(s)
90+ return datetime.fromisoformat(s).astimezone(UTC).date()
91+ except ValueError as exc:
92+ raise ApiError(400, f"{name} must be YYYY-MM-DD, got {v!r}") from exc
93+
94+
95+def parse_ts(v: str | None, name: str = "timestamp") -> datetime | None:
96+ """ISO-8601 (date or datetime) → aware UTC `datetime`."""
97+ if v is None or v == "":
98+ return None
99+ s = v.strip().replace("Z", "+00:00")
100+ try:
101+ dt = datetime.fromisoformat(s)
102+ except ValueError as exc:
103+ raise ApiError(400, f"{name} must be ISO-8601, got {v!r}") from exc
104+ if dt.tzinfo is None:
105+ dt = datetime.combine(dt.date(), dt.time() or dtime.min, UTC)
106+ return dt.astimezone(UTC)
107+
108+
109+def day_bounds(d: date) -> tuple[datetime, datetime]:
110+ start = datetime.combine(d, dtime.min, UTC)
111+ return start, datetime.combine(d, dtime.max, UTC).replace(microsecond=999999)
112+
113+
114+def csv(v: str | None) -> list[str]:
115+ return [x.strip() for x in (v or "").split(",") if x.strip()]
116+
117+
118+def num_expr(path: str) -> str:
119+ """Safe numeric cast of a JSON text attribute (`e.attributes->>'x'`)."""
120+ return f"(case when {path} ~ '^-?[0-9]+(\\.[0-9]+)?$' then ({path})::double precision end)"
121+
122+
123+def attr_num(key: str, alias: str = "e") -> str:
124+ return num_expr(f"{alias}.attributes->>'{key}'")
125+
126+
127+def flip_order(order_sql: str, order: str) -> str:
128+ """Apply `order=asc|desc` to the primary sort key of a canned ORDER BY fragment (keeps `nulls last`)."""
129+ if order not in ("asc", "desc"):
130+ return order_sql
131+ head, sep, tail = order_sql.partition(",")
132+ if order == "asc" and " desc" in head:
133+ head = head.replace(" desc", " asc", 1)
134+ elif order == "desc" and " asc" in head:
135+ head = head.replace(" asc", " desc", 1)
136+ return head + sep + tail
137+
138+
139+# ------------------------------------------------------------------------------------------------------------------ response cache
140+
141+
142+def cache_key(request: Request) -> str:
143+ items = sorted(request.query_params.multi_items())
144+ qs = "&".join(f"{k}={v}" for k, v in items)
145+ return f"{request.url.path}?{qs}" if qs else request.url.path
146+
147+
148+def cached(ttl_s: int) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
149+ """Cache a GET route body in Redis (`aia:api:<path>?<sorted query>`). The route must accept `request: Request`."""
150+
151+ def deco(fn: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
152+ @wraps(fn)
153+ async def wrapper(*args: Any, **kwargs: Any) -> Any:
154+ request = kwargs.get("request") or next((a for a in args if isinstance(a, Request)), None)
155+ key = cache_key(request) if request is not None else None
156+ if key is not None:
157+ hit = await cache.cache_get(key)
158+ if hit is not None:
159+ return hit
160+ value = normalize(await fn(*args, **kwargs))
161+ if key is not None:
162+ await cache.cache_set(key, value, ttl_s)
163+ return value
164+
165+ wrapper.__signature__ = inspect.signature(fn) # type: ignore[attr-defined]
166+ return wrapper
167+
168+ return deco
169+
170+
171+# ------------------------------------------------------------------------------------------------------------------ admin auth
172+
173+
174+def require_admin(request: Request) -> None:
175+ expected = settings.admin_token
176+ if not expected:
177+ raise ApiError(503, "admin API disabled: AIA_ADMIN_TOKEN is not configured")
178+ token = request.headers.get("x-aia-admin-token") or ""
179+ if not hmac.compare_digest(token, expected):
180+ raise ApiError(401, "invalid or missing x-aia-admin-token")
181+
182+
183+# ------------------------------------------------------------------------------------------------------------------ rate limiting
184+
185+_LIMITS: dict[str, tuple[int, float]] = {"search": (60, 60.0), "views": (1, 1.0), "admin": (240, 60.0), "default": (600, 60.0)}
186+_hits: dict[tuple[str, str], deque[float]] = defaultdict(deque)
187+
188+
189+def client_ip(request: Request) -> str:
190+ fwd = request.headers.get("x-forwarded-for")
191+ if fwd:
192+ return fwd.split(",")[0].strip()
193+ return request.client.host if request.client else "unknown"
194+
195+
196+def rate_limit(bucket: str) -> Callable[[Request], None]:
197+ def dep(request: Request) -> None:
198+ n, window = _LIMITS.get(bucket, _LIMITS["default"])
199+ now = time.monotonic()
200+ q = _hits[(client_ip(request), bucket)]
201+ while q and q[0] <= now - window:
202+ q.popleft()
203+ if len(q) >= n:
204+ raise ApiError(429, f"rate limit exceeded for {bucket}: {n} per {int(window)} s")
205+ q.append(now)
206+ if len(_hits) > 20000:
207+ for k in [k for k, v in _hits.items() if not v or v[-1] < now - 120]:
208+ _hits.pop(k, None)
209+
210+ return dep
211+
212+
213+# ------------------------------------------------------------------------------------------------------------------ entity serialisers
214+
215+ENTITY_FIELDS = ("id", "entity_type", "canonical_name", "slug", "description", "status", "organization_id", "attributes", "quality", "counts",
216+ "first_seen_at", "last_seen_at", "updated_at", "organization_name", "organization_slug")
217+
218+
219+def entity_cols(alias: str = "e", prefix: str = "") -> str:
220+ """Select-list for an entity alias (+ its organization alias `<alias>o`) with an optional column prefix."""
221+ org = f"{alias}o"
222+ cols = [f"{alias}.{c} as {prefix}{c}" for c in ENTITY_FIELDS[:13]]
223+ cols += [f"{org}.canonical_name as {prefix}organization_name", f"{org}.slug as {prefix}organization_slug"]
224+ return ", ".join(cols)
225+
226+
227+def entity_join(alias: str, on: str) -> str:
228+ return f"join entities {alias} on {alias}.id = {on} left join entities {alias}o on {alias}o.id = {alias}.organization_id"
229+
230+
231+ENTITY_COLS = entity_cols("e")
232+ENTITY_FROM = "entities e left join entities eo on eo.id = e.organization_id"
233+
234+SUMMARY_ATTRS: dict[str, tuple[str, ...]] = {
235+ "model": ("family", "openness", "license", "parameter_count", "active_parameter_count", "context_length", "max_output_tokens", "modalities",
236+ "release_date", "status", "knowledge_cutoff", "hf_repo", "api_model_id"),
237+ "company": ("country", "founded", "website", "org_kind"),
238+ "organization": ("country", "founded", "website", "org_kind"),
239+ "lab": ("country", "founded", "website", "org_kind"),
240+ "university": ("country", "founded", "website", "org_kind"),
241+ "provider": ("website", "pricing_url"),
242+ "benchmark": ("category", "metric", "unit"),
243+ "hardware": ("kind", "memory_gb", "memory_bandwidth_gbs", "release_date", "manufacturer"),
244+ "framework": ("latest_version", "latest_release_at", "license", "language", "metric.stars"),
245+ "repository": ("latest_version", "latest_release_at", "license", "language", "metric.stars"),
246+ "paper": ("authors", "published_at", "arxiv_id", "primary_category"),
247+ "dataset": ("license", "modality"),
248+}
249+COMPANY_TYPES = ("company", "organization", "lab", "university")
250+TYPE_LABELS = {"model": "Models", "company": "Companies", "organization": "Organizations", "lab": "Labs", "university": "Universities",
251+ "provider": "Providers", "paper": "Papers", "benchmark": "Benchmarks", "hardware": "Hardware", "framework": "Frameworks",
252+ "dataset": "Datasets", "tool": "Tools", "repository": "Repositories", "regulation": "Regulation", "incident": "Incidents",
253+ "researcher": "Researchers", "agent": "Agents", "product": "Products", "runtime": "Runtimes", "conference": "Conferences"}
254+
255+
256+def summary_attributes(entity_type: str, attrs: dict[str, Any] | None) -> dict[str, Any]:
257+ attrs = attrs or {}
258+ keys = SUMMARY_ATTRS.get(entity_type)
259+ if keys is None:
260+ return {k: v for k, v in attrs.items() if isinstance(v, (int, float, bool)) or (isinstance(v, str) and len(v) <= 200)}
261+ out: dict[str, Any] = {}
262+ for k in keys:
263+ if k in attrs and attrs[k] not in (None, "", [], {}):
264+ v = attrs[k]
265+ out[k] = v[:5] if k == "authors" and isinstance(v, list) else v
266+ return out
267+
268+
269+def org_of(row: dict[str, Any], prefix: str = "") -> dict[str, Any] | None:
270+ oid = row.get(f"{prefix}organization_id")
271+ if not oid:
272+ return None
273+ return {"id": oid, "slug": row.get(f"{prefix}organization_slug"), "name": row.get(f"{prefix}organization_name")}
274+
275+
276+def entity_summary(row: dict[str, Any], prefix: str = "") -> dict[str, Any] | None:
277+ g = row.get
278+ if not g(f"{prefix}id"):
279+ return None
280+ etype = g(f"{prefix}entity_type") or ""
281+ desc = g(f"{prefix}description")
282+ return {"id": g(f"{prefix}id"), "entity_type": etype, "slug": g(f"{prefix}slug"), "name": g(f"{prefix}canonical_name"),
283+ "description": (desc[:280] if isinstance(desc, str) and len(desc) > 280 else desc), "status": g(f"{prefix}status"),
284+ "organization": org_of(row, prefix), "attributes": summary_attributes(etype, g(f"{prefix}attributes")),
285+ "quality": g(f"{prefix}quality") or {}, "counts": g(f"{prefix}counts") or {},
286+ "first_seen_at": g(f"{prefix}first_seen_at"), "last_seen_at": g(f"{prefix}last_seen_at"), "updated_at": g(f"{prefix}updated_at")}
287+
288+
289+def change_event(row: dict[str, Any], prefix: str = "e_") -> dict[str, Any]:
290+ return {"id": row["id"], "event_type": row["event_type"], "category": row["category"], "property": row.get("property"),
291+ "old_value": row.get("old_value"), "new_value": row.get("new_value"), "summary": row.get("summary"), "importance": row.get("importance"),
292+ "observed_at": row.get("observed_at"), "effective_at": row.get("effective_at"), "source_url": row.get("source_url"),
293+ "connector_name": row.get("connector_name"), "entity": entity_summary(row, prefix), "meta": row.get("meta") or {}}
294+
295+
296+EVENT_COLS = ("ev.id, ev.event_type, ev.category, ev.property, ev.old_value, ev.new_value, ev.summary, ev.importance, ev.observed_at, ev.effective_at, "
297+ "ev.source_url, ev.connector_name, ev.meta, " + entity_cols("e", "e_"))
298+EVENT_FROM = "change_events ev left join entities e on e.id = ev.entity_id left join entities eo on eo.id = e.organization_id"
299+
300+
301+def price_row(row: dict[str, Any]) -> dict[str, Any]:
302+ return {"id": row["id"], "model": entity_summary(row, "m_"), "provider": entity_summary(row, "p_"), "provider_model_id": row.get("provider_model_id"),
303+ "input_per_mtok": row.get("input_per_mtok"), "output_per_mtok": row.get("output_per_mtok"), "cached_input_per_mtok": row.get("cached_input_per_mtok"),
304+ "cache_write_per_mtok": row.get("cache_write_per_mtok"), "batch_input_per_mtok": row.get("batch_input_per_mtok"),
305+ "batch_output_per_mtok": row.get("batch_output_per_mtok"), "per_image": row.get("per_image"), "per_request": row.get("per_request"),
306+ "currency": row.get("currency"), "context_length": row.get("context_length"), "max_output_tokens": row.get("max_output_tokens"),
307+ "features": row.get("features") or {}, "observed_at": row.get("observed_at"), "valid_from": row.get("valid_from"), "valid_to": row.get("valid_to"),
308+ "source_url": row.get("source_url"), "tier": row.get("tier")}
309+
310+
311+PRICE_COLS = ("p.id, p.provider_model_id, p.input_per_mtok, p.output_per_mtok, p.cached_input_per_mtok, p.cache_write_per_mtok, p.batch_input_per_mtok, "
312+ "p.batch_output_per_mtok, p.per_image, p.per_request, p.currency, p.context_length, p.max_output_tokens, p.features, p.observed_at, p.valid_from, "
313+ "p.valid_to, p.source_url, p.tier, " + entity_cols("m", "m_") + ", " + entity_cols("pv", "p_"))
314+PRICE_FROM = "prices p " + entity_join("m", "p.model_id") + " " + entity_join("pv", "p.provider_id")
315+
316+
317+def result_row(row: dict[str, Any]) -> dict[str, Any]:
318+ return {"id": row["id"], "model": entity_summary(row, "m_"), "benchmark": entity_summary(row, "b_"), "score": row.get("score"), "metric": row.get("metric"),
319+ "unit": row.get("unit"), "higher_is_better": row.get("higher_is_better"), "config": row.get("config") or {}, "evaluated_at": row.get("evaluated_at"),
320+ "observed_at": row.get("observed_at"), "source_url": row.get("source_url"), "tier": row.get("tier"), "confidence": row.get("confidence"),
321+ "valid_to": row.get("valid_to")}
322+
323+
324+RESULT_COLS = ("r.id, r.score, r.metric, r.unit, r.higher_is_better, r.config, r.evaluated_at, r.observed_at, r.source_url, r.tier, r.confidence, r.valid_to, "
325+ + entity_cols("m", "m_") + ", " + entity_cols("b", "b_"))
326+RESULT_FROM = "benchmark_results r " + entity_join("m", "r.model_id") + " " + entity_join("b", "r.benchmark_id")
327+RESULT_ORDER = "case when r.higher_is_better then -r.score else r.score end, r.observed_at desc"
328+
329+
330+def claim_row(row: dict[str, Any]) -> dict[str, Any]:
331+ return {"id": row["id"], "property": row["property"], "value": row.get("value"), "unit": row.get("unit"), "tier": row.get("tier"),
332+ "confidence": row.get("confidence"), "status": row.get("status"), "extractor": row.get("extractor"), "observed_at": row.get("observed_at"),
333+ "effective_at": row.get("effective_at"), "valid_from": row.get("valid_from"), "valid_to": row.get("valid_to"), "source_url": row.get("source_url"),
334+ "source_name": row.get("source_name")}
335+
336+
337+CLAIM_COLS = ("c.id, c.property, c.value, c.unit, c.tier, c.confidence, c.status, c.extractor, c.observed_at, c.effective_at, c.valid_from, c.valid_to, "
338+ "c.source_url, s.name as source_name")
339+CLAIM_FROM = "claims c left join sources s on s.id = c.source_id"
340+
341+
342+# ------------------------------------------------------------------------------------------------------------------ entity resolution
343+
344+
345+async def resolve_entity(conn: AsyncConnection, slug_or_id: str, types: tuple[str, ...] | None = None) -> dict[str, Any]:
346+ """Slug first, then id; follows `merged_into`; 404 when absent or when the type does not match the mounted alias."""
347+ key = slug_or_id.strip()
348+ if not key or len(key) > 200:
349+ raise ApiError(404, "entity not found")
350+ row = await fetch_one(conn, f"select {ENTITY_COLS}, e.provenance, e.merged_into from {ENTITY_FROM} where e.slug = :k or e.id = :k "
351+ f"order by case when e.slug = :k then 0 else 1 end limit 1", k=key)
352+ hops = 0
353+ while row and row.get("merged_into") and hops < 5:
354+ row = await fetch_one(conn, f"select {ENTITY_COLS}, e.provenance, e.merged_into from {ENTITY_FROM} where e.id = :k", k=row["merged_into"])
355+ hops += 1
356+ if not row:
357+ raise ApiError(404, "entity not found")
358+ if types and row["entity_type"] not in types:
359+ raise ApiError(404, f"entity {key!r} is a {row['entity_type']}, not one of {', '.join(types)}")
360+ return row
361+
362+
363+async def resolve_id(conn: AsyncConnection, slug_or_id: str | None, types: tuple[str, ...] | None = None) -> str | None:
364+ if not slug_or_id:
365+ return None
366+ return (await resolve_entity(conn, slug_or_id, types))["id"]
367+
368+
369+__all__ = [
370+ "CLAIM_COLS",
371+ "CLAIM_FROM",
372+ "COMPANY_TYPES",
373+ "ENTITY_COLS",
374+ "ENTITY_FROM",
375+ "EVENT_COLS",
376+ "EVENT_FROM",
377+ "PAGINATION",
378+ "PRICE_COLS",
379+ "PRICE_FROM",
380+ "RESULT_COLS",
381+ "RESULT_FROM",
382+ "RESULT_ORDER",
383+ "TYPE_LABELS",
384+ "ApiError",
385+ "AtlasJSONResponse",
386+ "Pagination",
387+ "attr_num",
388+ "cache_key",
389+ "cached",
390+ "change_event",
391+ "claim_row",
392+ "client_ip",
393+ "csv",
394+ "day_bounds",
395+ "dumps",
396+ "entity_cols",
397+ "entity_join",
398+ "entity_summary",
399+ "flip_order",
400+ "normalize",
401+ "num_expr",
402+ "org_of",
403+ "page",
404+ "parse_date",
405+ "parse_ts",
406+ "price_row",
407+ "rate_limit",
408+ "require_admin",
409+ "resolve_entity",
410+ "resolve_id",
411+ "result_row",
412+]
added src/aiatlas/api/detail.py +258 −0
@@ -0,0 +1,258 @@
1+"""`EntityDetail` builder (docs/API.md): shared by `/entities/{slug}` and the type-scoped aliases."""
2+from __future__ import annotations
3+
4+import asyncio
5+from typing import Any
6+
7+from sqlalchemy.ext.asyncio import AsyncConnection
8+
9+from aiatlas.api.common import (
10+ COMPANY_TYPES,
11+ ENTITY_COLS,
12+ ENTITY_FROM,
13+ EVENT_COLS,
14+ EVENT_FROM,
15+ PRICE_COLS,
16+ PRICE_FROM,
17+ RESULT_COLS,
18+ RESULT_FROM,
19+ RESULT_ORDER,
20+ change_event,
21+ entity_summary,
22+ price_row,
23+ result_row,
24+)
25+from aiatlas.db import connection, fetch_all
26+from aiatlas.services import hardware_fit as hf
27+
28+LINEAGE_PREDICATES = ("derived_from", "fine_tuned_from", "distilled_from", "merged_from", "quantized_from")
29+RELATION_GROUP_LIMIT = 24
30+
31+
32+async def relations_grouped(conn: AsyncConnection, entity_id: str) -> list[dict[str, Any]]:
33+ rows = await fetch_all(conn, f"""
34+ with rel as (
35+ select r.predicate, 'out' as direction, r.object_id as other_id, r.observed_at from relations r where r.subject_id = :id and r.valid_to is null
36+ union all
37+ select r.predicate, 'in' as direction, r.subject_id as other_id, r.observed_at from relations r where r.object_id = :id and r.valid_to is null),
38+ ranked as (select rel.*, row_number() over (partition by predicate, direction order by observed_at desc) as rn,
39+ count(*) over (partition by predicate, direction) as total from rel)
40+ select k.predicate, k.direction, k.total, {ENTITY_COLS} from ranked k join entities e on e.id = k.other_id left join entities eo on eo.id = e.organization_id
41+ where k.rn <= :lim and e.merged_into is null order by k.predicate, k.direction, k.rn""", id=entity_id, lim=RELATION_GROUP_LIMIT)
42+ groups: dict[tuple[str, str], dict[str, Any]] = {}
43+ for r in rows:
44+ g = groups.setdefault((r["predicate"], r["direction"]), {"predicate": r["predicate"], "direction": r["direction"], "items": [], "total": int(r["total"])})
45+ g["items"].append(entity_summary(r))
46+ return list(groups.values())
47+
48+
49+async def sources_of(conn: AsyncConnection, entity_id: str, limit: int = 60) -> list[dict[str, Any]]:
50+ """Documents attached to the entity + documents whose snapshots back its claims (deduped by URL — `documents.url` is unique)."""
51+ rows = await fetch_all(conn, """
52+ with docs as (select d.id from documents d where d.entity_id = :id
53+ union select s.document_id from claims c join snapshots s on s.id = c.snapshot_id where c.entity_id = :id
54+ union select s.document_id from relations r join snapshots s on s.id = r.snapshot_id where r.subject_id = :id or r.object_id = :id)
55+ select d.url, d.doc_type, d.title, greatest(d.last_fetched_at, d.last_changed_at) as last_observed_at, s.id as source_id, s.name as source_name,
56+ s.domain, s.tier, (select count(*) from snapshots x where x.document_id = d.id) as snapshots
57+ from docs join documents d on d.id = docs.id left join sources s on s.id = d.source_id
58+ order by s.tier nulls last, last_observed_at desc nulls last limit :lim""", id=entity_id, lim=limit)
59+ return [{"source_id": r["source_id"], "source_name": r["source_name"], "domain": r["domain"], "tier": r["tier"], "url": r["url"], "doc_type": r["doc_type"],
60+ "title": r.get("title"), "last_observed_at": r["last_observed_at"], "snapshots": int(r["snapshots"] or 0)} for r in rows]
61+
62+
63+def _scope_sql(is_org: bool) -> str:
64+ if is_org:
65+ return ("(ev.entity_id = :id or ev.entity_id in (select id from entities where organization_id = :id union "
66+ "select object_id from relations where subject_id = :id and predicate in ('develops','owns','operates','published') and valid_to is null))")
67+ return "ev.entity_id = :id"
68+
69+
70+async def timeline_of(conn: AsyncConnection, entity_id: str, entity_type: str, *, limit: int = 30, before: Any = None,
71+ include_documents: bool = False) -> list[dict[str, Any]]:
72+ where = [_scope_sql(entity_type in COMPANY_TYPES)]
73+ params: dict[str, Any] = {"id": entity_id, "lim": limit}
74+ if before is not None:
75+ where.append("ev.observed_at < :before")
76+ params["before"] = before
77+ if not include_documents:
78+ where.append("ev.event_type <> 'DOCUMENT_CHANGED'")
79+ rows = await fetch_all(conn, f"select {EVENT_COLS} from {EVENT_FROM} where {' and '.join(where)} order by ev.observed_at desc, ev.id desc limit :lim", **params)
80+ return [change_event(r) for r in rows]
81+
82+
83+async def prices_of_model(conn: AsyncConnection, model_id: str, *, current_only: bool) -> list[dict[str, Any]]:
84+ cond = "and p.valid_to is null" if current_only else ""
85+ order = "p.input_per_mtok nulls last, pv.canonical_name" if current_only else "p.valid_from, p.id"
86+ rows = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where p.model_id = :id {cond} order by {order} limit 500", id=model_id)
87+ return [price_row(r) for r in rows]
88+
89+
90+async def prices_of_provider(conn: AsyncConnection, provider_id: str) -> list[dict[str, Any]]:
91+ rows = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where p.provider_id = :id and p.valid_to is null order by m.canonical_name limit 500", id=provider_id)
92+ return [price_row(r) for r in rows]
93+
94+
95+async def results_of_model(conn: AsyncConnection, model_id: str) -> list[dict[str, Any]]:
96+ rows = await fetch_all(conn, f"select {RESULT_COLS} from {RESULT_FROM} where r.model_id = :id and r.valid_to is null order by b.canonical_name, r.observed_at desc limit 300", id=model_id)
97+ return [result_row(r) for r in rows]
98+
99+
100+async def leaderboard(conn: AsyncConnection, benchmark_id: str, *, limit: int = 100, offset: int = 0, config: str | None = None, history: bool = False) -> list[dict[str, Any]]:
101+ where = ["r.benchmark_id = :id"]
102+ params: dict[str, Any] = {"id": benchmark_id, "lim": limit, "off": offset}
103+ if not history:
104+ where.append("r.valid_to is null")
105+ if config:
106+ where.append("r.config::text ilike :cfg")
107+ params["cfg"] = f"%{config}%"
108+ rows = await fetch_all(conn, f"select {RESULT_COLS} from {RESULT_FROM} where {' and '.join(where)} order by {RESULT_ORDER} limit :lim offset :off", **params)
109+ return [result_row(r) for r in rows]
110+
111+
112+async def lineage_of(conn: AsyncConnection, model_id: str) -> dict[str, list[dict[str, Any]]]:
113+ preds = list(LINEAGE_PREDICATES)
114+ desc_preds = [p for p in preds if p != "quantized_from"]
115+ ancestors = await fetch_all(conn, f"""
116+ with recursive up as (
117+ select r.object_id as id, 1 as depth from relations r where r.subject_id = :id and r.valid_to is null and r.predicate = any(cast(:preds as text[]))
118+ union
119+ select r.object_id, up.depth + 1 from up join relations r on r.subject_id = up.id and r.valid_to is null and r.predicate = any(cast(:preds as text[])) where up.depth < 3)
120+ select distinct on (e.id) up.depth, {ENTITY_COLS} from up join entities e on e.id = up.id left join entities eo on eo.id = e.organization_id
121+ where e.id <> :id and e.merged_into is null order by e.id, up.depth""", id=model_id, preds=preds)
122+ descendants = await fetch_all(conn, f"""
123+ with recursive down as (
124+ select r.subject_id as id, 1 as depth from relations r where r.object_id = :id and r.valid_to is null and r.predicate = any(cast(:preds as text[]))
125+ union
126+ select r.subject_id, down.depth + 1 from down join relations r on r.object_id = down.id and r.valid_to is null and r.predicate = any(cast(:preds as text[])) where down.depth < 3)
127+ select distinct on (e.id) down.depth, {ENTITY_COLS} from down join entities e on e.id = down.id left join entities eo on eo.id = e.organization_id
128+ where e.id <> :id and e.merged_into is null order by e.id, down.depth""", id=model_id, preds=desc_preds)
129+ quants = await fetch_all(conn, f"""select {ENTITY_COLS} from relations r join entities e on e.id = r.subject_id left join entities eo on eo.id = e.organization_id
130+ where r.object_id = :id and r.predicate = 'quantized_from' and r.valid_to is null and e.merged_into is null
131+ order by e.canonical_name limit 100""", id=model_id)
132+ key = lambda r: (r.get("depth", 0), r["canonical_name"] or "")
133+ return {"ancestors": [entity_summary(r) for r in sorted(ancestors, key=key)], "descendants": [entity_summary(r) for r in sorted(descendants, key=key)],
134+ "quantizations": [entity_summary(r) for r in quants]}
135+
136+
137+async def providers_of_model(conn: AsyncConnection, model_id: str) -> list[dict[str, Any]]:
138+ rows = await fetch_all(conn, f"""
139+ with ids as (select r.object_id as id from relations r where r.subject_id = :id and r.predicate = 'available_through' and r.valid_to is null
140+ union select p.provider_id from prices p where p.model_id = :id and p.valid_to is null)
141+ select {ENTITY_COLS} from ids join entities e on e.id = ids.id left join entities eo on eo.id = e.organization_id
142+ where e.merged_into is null order by e.canonical_name limit 100""", id=model_id)
143+ return [entity_summary(r) for r in rows]
144+
145+
146+async def hardware_fit_of_model(conn: AsyncConnection, attrs: dict[str, Any]) -> list[dict[str, Any]] | None:
147+ params = hf.parameter_count(attrs)
148+ if params is None:
149+ return None
150+ context = 8192
151+ rows = await fetch_all(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.entity_type = 'hardware' and e.merged_into is null and e.attributes ? 'memory_gb' order by e.canonical_name limit 200")
152+ out: list[dict[str, Any]] = []
153+ for r in rows:
154+ mem = hf.hardware_memory_gb(r["attributes"])
155+ if mem is None:
156+ continue
157+ f = hf.fit(params, mem, "4bit", context)
158+ out.append({"hardware": entity_summary(r), "memory_gb": mem, **f})
159+ out.sort(key=lambda x: (not x["fits"], -x["headroom_gb"]))
160+ return out
161+
162+
163+async def related_by_type(conn: AsyncConnection, entity_id: str, etype: str, *, limit: int = 50, offset: int = 0, include_org_children: bool = False) -> tuple[list[dict[str, Any]], int]:
164+ """Entities of `etype` linked to `entity_id` by any live relation (either direction) — optionally also those whose organization is `entity_id`."""
165+ org_sql = "union select id from entities where organization_id = :id" if include_org_children else ""
166+ rows = await fetch_all(conn, f"""
167+ with ids as (select r.object_id as id from relations r where r.subject_id = :id and r.valid_to is null
168+ union select r.subject_id from relations r where r.object_id = :id and r.valid_to is null {org_sql})
169+ select count(*) over () as total, {ENTITY_COLS} from ids join entities e on e.id = ids.id left join entities eo on eo.id = e.organization_id
170+ where e.entity_type = :t and e.merged_into is null order by coalesce(e.attributes->>'release_date', e.attributes->>'published_at', '') desc, e.updated_at desc
171+ limit :lim offset :off""", id=entity_id, t=etype, lim=limit, off=offset)
172+ return [entity_summary(r) for r in rows], int(rows[0]["total"]) if rows else 0
173+
174+
175+async def entity_detail(row: dict[str, Any]) -> dict[str, Any]:
176+ """Full detail. Blocks run on separate pooled connections concurrently (read-only, so no transaction needed)."""
177+ eid, etype = row["id"], row["entity_type"]
178+
179+ async def one(fn: Any, *args: Any, **kw: Any) -> Any:
180+ async with connection() as conn:
181+ return await fn(conn, *args, **kw)
182+
183+ async def base(conn: AsyncConnection) -> dict[str, Any]:
184+ aliases = await fetch_all(conn, "select alias from entity_aliases where entity_id = :id order by kind, alias limit 200", id=eid)
185+ idents = await fetch_all(conn, "select scheme, value from entity_identifiers where entity_id = :id order by scheme, value limit 200", id=eid)
186+ return {"aliases": [a["alias"] for a in aliases], "identifiers": [{"scheme": i["scheme"], "value": i["value"]} for i in idents]}
187+
188+ tasks: dict[str, Any] = {"base": one(base), "relations": one(relations_grouped, eid), "sources": one(sources_of, eid), "timeline": one(timeline_of, eid, etype)}
189+ if etype == "model":
190+ tasks["prices"] = one(prices_of_model, eid, current_only=True)
191+ tasks["price_history"] = one(prices_of_model, eid, current_only=False)
192+ tasks["results"] = one(results_of_model, eid)
193+ tasks["lineage"] = one(lineage_of, eid)
194+ tasks["providers"] = one(providers_of_model, eid)
195+ tasks["hardware_fit"] = one(hardware_fit_of_model, row["attributes"] or {})
196+ tasks["papers"] = one(related_by_type, eid, "paper", limit=24)
197+ tasks["repositories"] = one(related_by_type, eid, "repository", limit=24)
198+ elif etype in COMPANY_TYPES:
199+ tasks["models"] = one(related_by_type, eid, "model", limit=50, include_org_children=True)
200+ tasks["papers"] = one(related_by_type, eid, "paper", limit=24, include_org_children=True)
201+ tasks["repositories"] = one(related_by_type, eid, "repository", limit=24, include_org_children=True)
202+ elif etype == "provider":
203+ tasks["prices"] = one(prices_of_provider, eid)
204+ tasks["models"] = one(_provider_models, eid)
205+ elif etype == "benchmark":
206+ tasks["results"] = one(leaderboard, eid, limit=100)
207+ elif etype == "hardware":
208+ tasks["models"] = one(related_by_type, eid, "model", limit=50)
209+ elif etype == "framework":
210+ tasks["repositories"] = one(related_by_type, eid, "repository", limit=24)
211+
212+ keys = list(tasks)
213+ values = await asyncio.gather(*(tasks[k] for k in keys))
214+ blocks = dict(zip(keys, values, strict=True))
215+
216+ detail = entity_summary(row) or {}
217+ detail["attributes"] = row.get("attributes") or {}
218+ detail["provenance"] = row.get("provenance") or {}
219+ detail.update(blocks.pop("base"))
220+ detail["relations"] = blocks.pop("relations")
221+ detail["sources"] = blocks.pop("sources")
222+ detail["timeline"] = blocks.pop("timeline")
223+ for k, v in blocks.items():
224+ if k in ("models",) and isinstance(v, tuple):
225+ items, total = v
226+ detail[k] = {"items": items, "total": total, "limit": 50, "offset": 0}
227+ elif k in ("papers", "repositories") and isinstance(v, tuple):
228+ detail[k] = v[0]
229+ elif v is not None:
230+ detail[k] = v
231+ if etype == "model" and "hardware_fit" in detail:
232+ detail["hardware_fit_assumptions"] = hf.ASSUMPTIONS
233+ return detail
234+
235+
236+async def _provider_models(conn: AsyncConnection, provider_id: str) -> tuple[list[dict[str, Any]], int]:
237+ rows = await fetch_all(conn, f"""
238+ with ids as (select p.model_id as id from prices p where p.provider_id = :id and p.valid_to is null
239+ union select r.subject_id from relations r where r.object_id = :id and r.predicate = 'available_through' and r.valid_to is null)
240+ select count(*) over () as total, {ENTITY_COLS} from ids join entities e on e.id = ids.id left join entities eo on eo.id = e.organization_id
241+ where e.merged_into is null order by e.canonical_name limit 50""", id=provider_id)
242+ return [entity_summary(r) for r in rows], int(rows[0]["total"]) if rows else 0
243+
244+
245+__all__ = [
246+ "entity_detail",
247+ "hardware_fit_of_model",
248+ "leaderboard",
249+ "lineage_of",
250+ "prices_of_model",
251+ "prices_of_provider",
252+ "providers_of_model",
253+ "related_by_type",
254+ "relations_grouped",
255+ "results_of_model",
256+ "sources_of",
257+ "timeline_of",
258+]
added src/aiatlas/api/main.py +137 −0
@@ -0,0 +1,137 @@
1+"""FastAPI application — `/api/v1`. Loopback-only in production (Next.js proxies `/api/v1/*`); docs at `/api/v1/docs`."""
2+from __future__ import annotations
3+
4+import asyncio
5+import logging
6+from contextlib import asynccontextmanager
7+from datetime import UTC, datetime
8+from typing import Any
9+
10+from fastapi import FastAPI, Request
11+from fastapi.exceptions import RequestValidationError
12+from fastapi.middleware.cors import CORSMiddleware
13+from fastapi.middleware.gzip import GZipMiddleware
14+from starlette.exceptions import HTTPException as StarletteHTTPException
15+
16+import aiatlas
17+from aiatlas.api.common import AtlasJSONResponse
18+from aiatlas.api.routers import (
19+ admin,
20+ benchmarks,
21+ changes,
22+ companies,
23+ compare,
24+ diff,
25+ entities,
26+ explore,
27+ hardware,
28+ misc,
29+ models,
30+ papers,
31+ prices,
32+ providers,
33+ search,
34+ sources,
35+ stats,
36+ timeline,
37+)
38+from aiatlas.config import settings
39+from aiatlas.db import connection, dispose, fetch_val
40+from aiatlas.logging import setup_logging
41+from aiatlas.services import cache
42+from aiatlas.services.llm import gateway
43+
44+log = logging.getLogger("aiatlas.api")
45+
46+
47+@asynccontextmanager
48+async def lifespan(app: FastAPI): # type: ignore[no-untyped-def]
49+ setup_logging(service="aia-api")
50+ settings.ensure_dirs()
51+ log.info("api started", extra={"version": aiatlas.__version__, "env": settings.app_env, "port": settings.api_port})
52+ yield
53+ await cache.close()
54+ await dispose()
55+
56+
57+app = FastAPI(title="AI Atlas API", version=aiatlas.__version__, lifespan=lifespan, default_response_class=AtlasJSONResponse,
58+ docs_url="/api/v1/docs", redoc_url=None, openapi_url="/api/v1/openapi.json",
59+ description="The global intelligence layer for artificial intelligence: models, companies, papers, providers, prices, benchmarks, hardware — "
60+ "with provenance and history on every fact. Contract: docs/API.md.")
61+
62+app.add_middleware(GZipMiddleware, minimum_size=1024)
63+_origins = {settings.site_url, "https://www.ai-atlas.co", "https://ai-atlas.co", "http://localhost:8320", "http://127.0.0.1:8320", "http://localhost:8330", "http://127.0.0.1:8330"}
64+app.add_middleware(CORSMiddleware, allow_origins=sorted(o for o in _origins if o), allow_methods=["GET", "POST", "PATCH", "OPTIONS"],
65+ allow_headers=["*"], max_age=600)
66+
67+
68+@app.middleware("http")
69+async def security_headers(request: Request, call_next): # type: ignore[no-untyped-def]
70+ try:
71+ response = await call_next(request)
72+ except Exception:
73+ log.exception("unhandled error", extra={"route": request.url.path})
74+ return AtlasJSONResponse({"detail": "internal server error"}, status_code=500)
75+ response.headers["x-content-type-options"] = "nosniff"
76+ response.headers["referrer-policy"] = "strict-origin-when-cross-origin"
77+ return response
78+
79+
80+@app.exception_handler(StarletteHTTPException)
81+async def http_error(request: Request, exc: StarletteHTTPException): # type: ignore[no-untyped-def]
82+ detail = exc.detail if isinstance(exc.detail, str) else str(exc.detail)
83+ return AtlasJSONResponse({"detail": detail}, status_code=exc.status_code, headers=dict(exc.headers or {}))
84+
85+
86+@app.exception_handler(RequestValidationError)
87+async def validation_error(request: Request, exc: RequestValidationError): # type: ignore[no-untyped-def]
88+ errs = exc.errors()[:5]
89+ msg = "; ".join(f"{'.'.join(str(x) for x in e.get('loc', []) if x not in ('query', 'body', 'path'))}: {e.get('msg')}" for e in errs) or "invalid request"
90+ return AtlasJSONResponse({"detail": msg, "errors": [{"loc": e.get("loc"), "msg": e.get("msg"), "type": e.get("type")} for e in errs]}, status_code=422)
91+
92+
93+@app.exception_handler(Exception)
94+async def unhandled(request: Request, exc: Exception): # type: ignore[no-untyped-def]
95+ log.exception("unhandled error", extra={"route": request.url.path})
96+ return AtlasJSONResponse({"detail": "internal server error"}, status_code=500)
97+
98+
99+async def _health() -> dict[str, Any]:
100+ db_ok = redis_ok = False
101+ try:
102+ async with connection() as conn:
103+ db_ok = (await asyncio.wait_for(fetch_val(conn, "select 1"), timeout=3)) == 1
104+ except Exception: # noqa: BLE001
105+ db_ok = False
106+ try:
107+ redis_ok = bool(await asyncio.wait_for(cache.redis().ping(), timeout=2))
108+ except Exception: # noqa: BLE001
109+ redis_ok = False
110+ llm: dict[str, Any] = {"available": gateway.available}
111+ if gateway.available:
112+ cached = await cache.cache_get("health:llm")
113+ if cached is None:
114+ try:
115+ cached = {"reachable": bool(await asyncio.wait_for(gateway.engine.health(), timeout=2.5))}
116+ except Exception: # noqa: BLE001
117+ cached = {"reachable": False}
118+ await cache.cache_set("health:llm", cached, 300)
119+ llm.update(cached)
120+ return {"status": "ok" if db_ok else "degraded", "version": aiatlas.__version__, "db": db_ok, "redis": redis_ok, "llm": llm, "time": datetime.now(UTC)}
121+
122+
123+@app.get("/health", tags=["health"])
124+async def health_root() -> dict[str, Any]:
125+ return await _health()
126+
127+
128+@app.get("/api/v1/health", tags=["health"])
129+async def health_v1() -> dict[str, Any]:
130+ return await _health()
131+
132+
133+# Order matters only where a literal path and a `{param}` path share a prefix — literal routes live in the same router and are declared first.
134+for r in (stats, search, models, companies, papers, providers, prices, benchmarks, hardware, explore, changes, timeline, compare, diff, sources, misc, entities, admin):
135+ app.include_router(r.router)
136+
137+__all__ = ["app"]
added src/aiatlas/api/routers/__init__.py +1 −0
@@ -0,0 +1 @@
1+"""One router per docs/API.md group, all mounted under `/api/v1`."""
added src/aiatlas/api/routers/admin.py +539 −0
@@ -0,0 +1,539 @@
1+"""Admin API (`x-aia-admin-token`): overview, connectors, runs/errors, documents & snapshots (cleaned text only), queues, LLM
2+accounting, review queue with entity merging, duplicates, curation, infrastructure, cache. Never returns raw archive paths."""
3+from __future__ import annotations
4+
5+import platform
6+import socket
7+import time
8+from datetime import UTC, datetime
9+from typing import Any
10+
11+from fastapi import APIRouter, Depends, Query
12+from pydantic import BaseModel, Field
13+
14+from aiatlas.api.common import PAGINATION, ApiError, Pagination, page, rate_limit, require_admin
15+from aiatlas.config import settings
16+from aiatlas.connectors import registry as connector_registry
17+from aiatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction
18+from aiatlas.ids import new_id
19+from aiatlas.sdk import archive
20+from aiatlas.services import cache
21+from aiatlas.services.jobs import enqueue, queue_depth
22+from aiatlas.services.llm import gateway
23+from aiatlas.services.merge import merge_entities
24+from aiatlas.services.stats import live_counts
25+
26+router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin), Depends(rate_limit("admin"))])
27+_STARTED = time.time()
28+SNAPSHOT_TEXT_LIMIT = 20 * 1024
29+HIDDEN = ("raw_path", "text_path")
30+
31+
32+def _public(row: dict[str, Any]) -> dict[str, Any]:
33+ return {k: v for k, v in row.items() if k not in HIDDEN}
34+
35+
36+# ------------------------------------------------------------------------------------------------------------------ overview
37+
38+
39+@router.get("/overview")
40+async def overview() -> dict[str, Any]:
41+ async with connection() as conn:
42+ stats = await live_counts(conn)
43+ queue = await queue_depth(conn)
44+ health = await fetch_all(conn, "select health, count(*) as n from connectors group by 1")
45+ errors = await fetch_val(conn, "select count(*) from connector_errors where created_at > now() - interval '24 hours'")
46+ llm = await fetch_one(conn, """select count(*) as jobs_24h, coalesce(sum(input_tokens), 0) + coalesce(sum(output_tokens), 0) as tokens_24h,
47+ count(*) filter (where status <> 'ok') as failed_24h from llm_jobs where created_at > now() - interval '24 hours'""")
48+ by_stage = await fetch_all(conn, "select stage, status, count(*) as n from llm_jobs where created_at > now() - interval '24 hours' group by 1, 2 order by 1, 2")
49+ recent_runs = await fetch_all(conn, "select * from connector_runs order by started_at desc limit 12")
50+ review_kinds = await fetch_all(conn, "select kind, count(*) as n from review_queue where status = 'pending' group by 1")
51+ counts = {h["health"]: int(h["n"]) for h in health}
52+ return {"stats": stats, "queue": queue, "heartbeats": await cache.heartbeats(),
53+ "connectors": {k: counts.get(k, 0) for k in ("ok", "degraded", "failing", "disabled", "unknown")}, "review_pending": stats.get("review_pending", 0),
54+ "review_by_kind": {r["kind"]: int(r["n"]) for r in review_kinds}, "recent_errors": int(errors or 0),
55+ "llm": {"available": gateway.available, "jobs_24h": int(llm["jobs_24h"] or 0), "tokens_24h": int(llm["tokens_24h"] or 0), "failed_24h": int(llm["failed_24h"] or 0),
56+ "by_stage": [{**r, "n": int(r["n"])} for r in by_stage]},
57+ "recent_runs": recent_runs, "archive": archive.archive_size(), "computed_at": datetime.now(UTC)}
58+
59+
60+# ------------------------------------------------------------------------------------------------------------------ connectors
61+
62+
63+@router.get("/connectors")
64+async def connectors() -> dict[str, Any]:
65+ async with connection() as conn:
66+ rows = await fetch_all(conn, """
67+ select c.*, s.key as source_key, s.name as source_name, s.tier as source_tier, s.domain as source_domain,
68+ (select row_to_json(r) from connector_runs r where r.connector_name = c.name order by r.started_at desc limit 1) as last_run,
69+ (select count(*) from documents d where d.connector_name = c.name) as documents,
70+ (select count(*) from snapshots x join documents d on d.id = x.document_id where d.connector_name = c.name) as snapshots,
71+ (select count(*) from connector_errors e where e.connector_name = c.name and e.created_at > now() - interval '7 days') as errors_7d
72+ from connectors c left join sources s on s.id = c.source_id order by c.priority, c.name""")
73+ known = connector_registry()
74+ return {"items": [{**r, "documents": int(r["documents"]), "snapshots": int(r["snapshots"]), "errors_7d": int(r["errors_7d"]), "in_code": r["name"] in known,
75+ "run_now_pending": await _run_now_pending(r["name"])} for r in rows],
76+ "unregistered_in_db": sorted(set(known) - {r["name"] for r in rows})}
77+
78+
79+async def _run_now_pending(name: str) -> bool:
80+ try:
81+ return bool(await cache.redis().exists(f"aia:run-now:{name}"))
82+ except Exception: # noqa: BLE001
83+ return False
84+
85+
86+class RunBody(BaseModel):
87+ force: bool = False
88+
89+
90+@router.post("/connectors/{name}/run")
91+async def run_connector(name: str, body: RunBody | None = None) -> dict[str, Any]:
92+ async with connection() as conn:
93+ exists = await fetch_one(conn, "select name from connectors where name = :n", n=name)
94+ if not exists and name not in connector_registry():
95+ raise ApiError(404, f"unknown connector {name!r}")
96+ try:
97+ await cache.redis().set(f"aia:run-now:{name}", b"force" if (body and body.force) else b"1", ex=6 * 3600)
98+ except Exception as exc:
99+ raise ApiError(503, f"redis unavailable: {exc.__class__.__name__}") from exc
100+ return {"queued": True, "connector": name, "force": bool(body and body.force), "note": "consumed by the scheduler tick (aia:run-now:<name>)"}
101+
102+
103+class ConnectorPatch(BaseModel):
104+ enabled: bool | None = None
105+ interval_seconds: int | None = Field(None, ge=60, le=30 * 86400)
106+ priority: int | None = Field(None, ge=0, le=9)
107+
108+
109+@router.patch("/connectors/{name}")
110+async def patch_connector(name: str, body: ConnectorPatch) -> dict[str, Any]:
111+ sets, params = [], {"n": name}
112+ if body.enabled is not None:
113+ sets.append("enabled = :enabled")
114+ sets.append("health = case when :enabled then (case when health = 'disabled' then 'unknown' else health end) else 'disabled' end")
115+ sets.append("circuit_open_until = case when :enabled then null else circuit_open_until end")
116+ params["enabled"] = body.enabled
117+ if body.interval_seconds is not None:
118+ sets.append("interval_seconds = :iv")
119+ sets.append("next_run_at = least(coalesce(next_run_at, now()), coalesce(last_success_at, now()) + make_interval(secs => :iv))")
120+ params["iv"] = body.interval_seconds
121+ if body.priority is not None:
122+ sets.append("priority = :pr")
123+ params["pr"] = body.priority
124+ if not sets:
125+ raise ApiError(400, "nothing to update")
126+ async with transaction() as conn:
127+ row = await fetch_one(conn, f"update connectors set {', '.join(sets)}, updated_at = now() where name = :n returning *", **params)
128+ if not row:
129+ raise ApiError(404, f"unknown connector {name!r}")
130+ return row
131+
132+
133+@router.get("/runs")
134+async def runs(connector: str | None = None, status: str | None = None, limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0)) -> dict[str, Any]:
135+ where = ["true"]
136+ params: dict[str, Any] = {"lim": limit, "off": offset}
137+ if connector:
138+ where.append("connector_name = :c")
139+ params["c"] = connector
140+ if status:
141+ where.append("status = :s")
142+ params["s"] = status
143+ async with connection() as conn:
144+ rows = await fetch_all(conn, f"select * from connector_runs where {' and '.join(where)} order by started_at desc limit :lim offset :off", **params)
145+ total = await fetch_val(conn, f"select count(*) from connector_runs where {' and '.join(where)}", **{k: v for k, v in params.items() if k not in ('lim', 'off')})
146+ return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset}
147+
148+
149+@router.get("/errors")
150+async def errors(connector: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:
151+ where = "connector_name = :c" if connector else "true"
152+ async with connection() as conn:
153+ rows = await fetch_all(conn, f"select * from connector_errors where {where} order by created_at desc limit :lim offset :off", c=connector, lim=limit, off=offset)
154+ total = await fetch_val(conn, f"select count(*) from connector_errors where {where}", c=connector)
155+ return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset}
156+
157+
158+# ------------------------------------------------------------------------------------------------------------------ documents & snapshots
159+
160+
161+@router.get("/documents")
162+async def documents(connector: str | None = None, status: str | None = None, q: str | None = Query(None, max_length=300), entity: str | None = None,
163+ needs_llm: int | None = Query(None, ge=0, le=1), p: Pagination = PAGINATION) -> dict[str, Any]:
164+ where = ["true"]
165+ params: dict[str, Any] = {}
166+ if connector:
167+ where.append("d.connector_name = :c")
168+ params["c"] = connector
169+ if status:
170+ where.append("d.status = :s")
171+ params["s"] = status
172+ if q:
173+ where.append("(d.url ilike :q or d.title ilike :q)")
174+ params["q"] = f"%{q}%"
175+ if entity:
176+ where.append("(d.entity_id = :e or e.slug = :e)")
177+ params["e"] = entity
178+ if needs_llm is not None:
179+ where.append("d.needs_llm = :nl")
180+ params["nl"] = bool(needs_llm)
181+ where_sql = " and ".join(where)
182+ async with connection() as conn:
183+ rows = await fetch_all(conn, f"""select d.*, e.slug as entity_slug, e.canonical_name as entity_name, e.entity_type,
184+ (select count(*) from snapshots x where x.document_id = d.id) as snapshots,
185+ (select x.processing_status from snapshots x where x.document_id = d.id order by x.observed_at desc limit 1) as last_processing_status
186+ from documents d left join entities e on e.id = d.entity_id where {where_sql}
187+ order by d.last_fetched_at desc nulls last, d.first_seen_at desc limit :lim offset :off""", lim=p.limit, off=p.offset, **params)
188+ total = await fetch_val(conn, f"select count(*) from documents d left join entities e on e.id = d.entity_id where {where_sql}", **params)
189+ return page([{**r, "snapshots": int(r["snapshots"])} for r in rows], int(total or 0), p)
190+
191+
192+@router.get("/documents/{doc_id}")
193+async def document(doc_id: str) -> dict[str, Any]:
194+ async with connection() as conn:
195+ doc = await fetch_one(conn, """select d.*, e.slug as entity_slug, e.canonical_name as entity_name, e.entity_type, s.name as source_name, s.tier as source_tier
196+ from documents d left join entities e on e.id = d.entity_id left join sources s on s.id = d.source_id where d.id = :id or d.url = :id limit 1""", id=doc_id)
197+ if not doc:
198+ raise ApiError(404, "document not found")
199+ snaps = await fetch_all(conn, """select id, run_id, url, final_url, observed_at, http_status, content_type, content_hash, byte_size, text_hash, parser_version, connector_version,
200+ transport, changed, processing_status, created_at, structured is not null as has_structured, diff is not null as has_diff, text_path is not null as has_text
201+ from snapshots where document_id = :id order by observed_at desc limit 200""", id=doc["id"])
202+ claims = await fetch_val(conn, "select count(*) from claims c join snapshots s on s.id = c.snapshot_id where s.document_id = :id", id=doc["id"])
203+ return {**doc, "snapshots": snaps, "claims_from_document": int(claims or 0)}
204+
205+
206+@router.get("/snapshots/{snap_id}")
207+async def snapshot(snap_id: str, text_limit: int = Query(SNAPSHOT_TEXT_LIMIT, ge=0, le=SNAPSHOT_TEXT_LIMIT)) -> dict[str, Any]:
208+ async with connection() as conn:
209+ snap = await fetch_one(conn, """select s.*, d.url as document_url, d.doc_type, d.connector_name, d.entity_id, e.slug as entity_slug, e.canonical_name as entity_name
210+ from snapshots s join documents d on d.id = s.document_id left join entities e on e.id = d.entity_id where s.id = :id""", id=snap_id)
211+ if not snap:
212+ raise ApiError(404, "snapshot not found")
213+ claims = await fetch_all(conn, """select c.id, c.entity_id, e.slug as entity_slug, c.property, c.value, c.unit, c.status, c.confidence, c.extractor, c.observed_at
214+ from claims c left join entities e on e.id = c.entity_id where c.snapshot_id = :id order by e.slug, c.property limit 500""", id=snap_id)
215+ events = await fetch_all(conn, "select id, entity_id, event_type, category, property, summary, importance, observed_at from change_events where snapshot_id = :id order by observed_at desc limit 200", id=snap_id)
216+ llm = await fetch_all(conn, "select id, task_type, stage, model, status, input_tokens, output_tokens, duration_ms, error, created_at from llm_jobs where snapshot_id = :id order by created_at desc limit 50", id=snap_id)
217+ text, text_error, text_len = None, None, 0
218+ if snap.get("text_path") and text_limit:
219+ try:
220+ full = archive.load_text(snap["text_path"])
221+ text_len = len(full)
222+ text = full[:text_limit]
223+ except OSError as exc:
224+ text_error = f"cleaned text unavailable ({exc.__class__.__name__})"
225+ out = _public(snap)
226+ out.update({"text": text, "text_chars": text_len, "text_truncated": text_len > text_limit, "text_error": text_error, "has_raw": bool(snap.get("raw_path")),
227+ "claims": claims, "events": events, "llm_jobs": llm})
228+ return out
229+
230+
231+# ------------------------------------------------------------------------------------------------------------------ queues
232+
233+
234+@router.get("/jobs")
235+async def jobs(status: str | None = None, kind: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:
236+ where = ["true"]
237+ params: dict[str, Any] = {}
238+ if status:
239+ where.append("status = any(cast(:s as text[]))")
240+ params["s"] = [x.strip() for x in status.split(",") if x.strip()]
241+ if kind:
242+ where.append("kind = :k")
243+ params["k"] = kind
244+ where_sql = " and ".join(where)
245+ async with connection() as conn:
246+ rows = await fetch_all(conn, f"select * from jobs where {where_sql} order by case status when 'running' then 0 when 'queued' then 1 when 'failed' then 2 when 'dead' then 3 else 4 end, "
247+ f"priority, coalesce(finished_at, run_after) desc limit :lim offset :off", lim=limit, off=offset, **params)
248+ total = await fetch_val(conn, f"select count(*) from jobs where {where_sql}", **params)
249+ depth = await queue_depth(conn)
250+ return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "depth": depth}
251+
252+
253+@router.post("/jobs/{job_id}/retry")
254+async def retry_job(job_id: str) -> dict[str, Any]:
255+ async with transaction() as conn:
256+ row = await fetch_one(conn, """update jobs set status = 'queued', attempts = 0, error = null, locked_by = null, locked_at = null, finished_at = null, run_after = now()
257+ where id = :id and status in ('failed', 'dead', 'done') returning id, kind, status""", id=job_id)
258+ if not row:
259+ raise ApiError(404, "job not found or not retryable (must be failed, dead or done)")
260+ return {"ok": True, **row}
261+
262+
263+@router.post("/jobs/requeue-dead")
264+async def requeue_dead(kind: str | None = None) -> dict[str, Any]:
265+ where = "status = 'dead'" + (" and kind = :k" if kind else "")
266+ async with transaction() as conn:
267+ n = await fetch_val(conn, f"with u as (update jobs set status = 'queued', attempts = 0, error = null, locked_by = null, locked_at = null, finished_at = null, run_after = now() "
268+ f"where {where} returning 1) select count(*) from u", k=kind)
269+ return {"requeued": int(n or 0)}
270+
271+
272+@router.get("/llm-jobs")
273+async def llm_jobs(limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0), status: str | None = None, task: str | None = None) -> dict[str, Any]:
274+ where = ["true"]
275+ params: dict[str, Any] = {}
276+ if status:
277+ where.append("status = :s")
278+ params["s"] = status
279+ if task:
280+ where.append("task_type = :t")
281+ params["t"] = task
282+ where_sql = " and ".join(where)
283+ async with connection() as conn:
284+ rows = await fetch_all(conn, f"select id, job_id, task_type, stage, engine, model, node, schema_name, snapshot_id, entity_id, input_tokens, output_tokens, duration_ms, status, error, created_at "
285+ f"from llm_jobs where {where_sql} order by created_at desc limit :lim offset :off", lim=limit, off=offset, **params)
286+ total = await fetch_val(conn, f"select count(*) from llm_jobs where {where_sql}", **params)
287+ totals = await fetch_all(conn, """select stage, model, status, count(*) as n, coalesce(sum(input_tokens), 0) as input_tokens, coalesce(sum(output_tokens), 0) as output_tokens,
288+ coalesce(avg(duration_ms), 0)::int as avg_ms from llm_jobs group by 1, 2, 3 order by 1, 2, 3""")
289+ return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "totals": [{**t, "n": int(t["n"]), "input_tokens": int(t["input_tokens"]), "output_tokens": int(t["output_tokens"])} for t in totals]}
290+
291+
292+@router.get("/llm/health")
293+async def llm_health() -> dict[str, Any]:
294+ h = await gateway.health()
295+ h["base_url_configured"] = bool(settings.llm_base_url)
296+ return h
297+
298+
299+class LLMEnqueue(BaseModel):
300+ limit: int = Field(200, ge=1, le=5000)
301+ task: str | None = None
302+ connector: str | None = None
303+
304+
305+@router.post("/llm/enqueue")
306+async def llm_enqueue(body: LLMEnqueue | None = None) -> dict[str, Any]:
307+ body = body or LLMEnqueue()
308+ where = "s.processing_status in ('stored', 'extracted', 'llm_pending') and s.text_path is not null and (d.needs_llm or s.processing_status = 'llm_pending')"
309+ params: dict[str, Any] = {"lim": body.limit}
310+ if body.connector:
311+ where += " and d.connector_name = :c"
312+ params["c"] = body.connector
313+ queued = 0
314+ async with transaction() as conn:
315+ snaps = await fetch_all(conn, f"select s.id from snapshots s join documents d on d.id = s.document_id where {where} order by d.priority, s.observed_at desc limit :lim", **params)
316+ for s in snaps:
317+ payload = {"snapshot_id": s["id"], **({"task": body.task} if body.task else {})}
318+ if await enqueue(conn, "llm_extract", payload, priority=6, dedupe_key=f"llm:{s['id']}"):
319+ queued += 1
320+ if snaps:
321+ await execute(conn, "update snapshots set processing_status = 'llm_pending' where id = any(cast(:ids as text[])) and processing_status <> 'llm_pending'", ids=[s["id"] for s in snaps])
322+ return {"queued": queued, "candidates": len(snaps), "llm_available": gateway.available}
323+
324+
325+class ReprocessBody(BaseModel):
326+ connector: str
327+ url: str | None = None
328+
329+
330+@router.post("/reprocess")
331+async def reprocess(body: ReprocessBody) -> dict[str, Any]:
332+ if body.connector not in connector_registry():
333+ raise ApiError(404, f"unknown connector {body.connector!r}")
334+ async with transaction() as conn:
335+ jid = await enqueue(conn, "reprocess_snapshot", {"connector": body.connector, **({"url": body.url} if body.url else {})}, priority=3,
336+ dedupe_key=f"reprocess:{body.connector}:{body.url or '*'}")
337+ return {"queued": jid is not None, "job_id": jid}
338+
339+
340+# ------------------------------------------------------------------------------------------------------------------ review queue & curation
341+
342+
343+@router.get("/review")
344+async def review(status: str = "pending", kind: str | None = None, limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0)) -> dict[str, Any]:
345+ where = ["r.status = :s"] if status != "all" else ["true"]
346+ params: dict[str, Any] = {"s": status}
347+ if kind:
348+ where.append("r.kind = :k")
349+ params["k"] = kind
350+ where_sql = " and ".join(where)
351+ async with connection() as conn:
352+ rows = await fetch_all(conn, f"""select r.*, (select jsonb_agg(jsonb_build_object('id', e.id, 'slug', e.slug, 'name', e.canonical_name, 'entity_type', e.entity_type, 'status', e.status) order by e.slug)
353+ from entities e where e.id = any(r.entity_ids)) as entities
354+ from review_queue r where {where_sql} order by r.created_at desc limit :lim offset :off""", lim=limit, off=offset, **params)
355+ total = await fetch_val(conn, f"select count(*) from review_queue r where {where_sql}", **params)
356+ kinds = await fetch_all(conn, "select kind, status, count(*) as n from review_queue group by 1, 2 order by 1, 2")
357+ return {"items": rows, "total": int(total or 0), "limit": limit, "offset": offset, "by_kind": [{**k, "n": int(k["n"])} for k in kinds]}
358+
359+
360+class ReviewAction(BaseModel):
361+ action: str = Field(..., pattern="^(approve|reject|edit)$")
362+ resolution: dict[str, Any] | None = None
363+
364+
365+@router.post("/review/{review_id}")
366+async def review_action(review_id: str, body: ReviewAction) -> dict[str, Any]:
367+ async with transaction() as conn:
368+ item = await fetch_one(conn, "select * from review_queue where id = :id for update", id=review_id)
369+ if not item:
370+ raise ApiError(404, "review item not found")
371+ if item["status"] != "pending":
372+ raise ApiError(409, f"review item already {item['status']}")
373+ resolution: dict[str, Any] = dict(body.resolution or {})
374+ effect: dict[str, Any] | None = None
375+ if body.action == "approve" and item["kind"] == "merge_candidate":
376+ payload = item["payload"] or {}
377+ ids = list(item["entity_ids"] or [])
378+ source = resolution.get("source_id") or payload.get("source_id") or (ids[0] if len(ids) > 1 else None)
379+ target = resolution.get("target_id") or payload.get("target_id") or (ids[1] if len(ids) > 1 else None)
380+ if not source or not target:
381+ raise ApiError(400, "merge_candidate needs source_id and target_id (payload or resolution)")
382+ try:
383+ effect = await merge_entities(conn, source, target)
384+ except (ValueError, LookupError) as exc:
385+ raise ApiError(400, str(exc)) from exc
386+ resolution.update({"merged": effect})
387+ elif body.action == "approve" and item["kind"] == "conflict":
388+ keep = resolution.get("keep_claim_id")
389+ if keep:
390+ effect = await _accept_claim(conn, keep)
391+ resolution["accepted"] = effect
392+ status = {"approve": "approved", "reject": "rejected", "edit": "edited"}[body.action]
393+ await execute(conn, "update review_queue set status = :st, resolution = cast(:r as jsonb), resolved_at = now() where id = :id", st=status, r=jsonb(resolution), id=review_id)
394+ await cache.cache_invalidate()
395+ return {"ok": True, "id": review_id, "status": status, "effect": effect}
396+
397+
398+async def _accept_claim(conn: Any, claim_id: str) -> dict[str, Any]:
399+ """Conflict resolution: promote one claim to current, supersede the others for the same property."""
400+ c = await fetch_one(conn, "select * from claims where id = :id", id=claim_id)
401+ if not c:
402+ raise ApiError(404, "claim not found")
403+ await execute(conn, "update claims set status = 'superseded', valid_to = now() where entity_id = :e and property = :p and id <> :id and status in ('current', 'conflicting')",
404+ e=c["entity_id"], p=c["property"], id=claim_id)
405+ await execute(conn, "update claims set status = 'current', confidence = case when confidence = 'conflicted' then 'high' else confidence end, valid_to = null where id = :id", id=claim_id)
406+ prov = {"source_id": c["source_id"], "snapshot_id": c["snapshot_id"], "url": c["source_url"], "tier": c["tier"], "confidence": "high", "extractor": c["extractor"],
407+ "observed_at": c["observed_at"].isoformat() if c["observed_at"] else None, "resolved_by": "review"}
408+ await execute(conn, "update entities set attributes = attributes || jsonb_build_object(:p, cast(:v as jsonb)), provenance = provenance || jsonb_build_object(:p, cast(:pv as jsonb)), updated_at = now() where id = :e",
409+ p=c["property"], v=jsonb(c["value"]), pv=jsonb(prov), e=c["entity_id"])
410+ return {"claim_id": claim_id, "entity_id": c["entity_id"], "property": c["property"]}
411+
412+
413+@router.get("/entities/duplicates")
414+async def duplicates(type: str | None = Query(None, alias="type"), limit: int = Query(100, ge=1, le=500), threshold: float = Query(0.8, ge=0.3, le=1.0)) -> dict[str, Any]:
415+ where = "a.entity_type = :t" if type else "true"
416+ async with connection() as conn:
417+ await execute(conn, "select set_limit(:th)", th=threshold)
418+ rows = await fetch_all(conn, f"""
419+ select a.id as a_id, a.slug as a_slug, a.canonical_name as a_name, a.entity_type, a.first_seen_at as a_first_seen_at, oa.canonical_name as a_org,
420+ b.id as b_id, b.slug as b_slug, b.canonical_name as b_name, b.first_seen_at as b_first_seen_at, ob.canonical_name as b_org,
421+ similarity(a.canonical_name, b.canonical_name) as similarity,
422+ (select count(*) from claims c where c.entity_id = a.id and c.status = 'current') as a_claims,
423+ (select count(*) from claims c where c.entity_id = b.id and c.status = 'current') as b_claims
424+ from entities a join entities b on b.entity_type = a.entity_type and b.id > a.id and a.canonical_name % b.canonical_name
425+ left join entities oa on oa.id = a.organization_id left join entities ob on ob.id = b.organization_id
426+ where {where} and a.merged_into is null and b.merged_into is null and similarity(a.canonical_name, b.canonical_name) > :th
427+ order by similarity desc, a.canonical_name limit :lim""", t=type, th=threshold, lim=limit)
428+ return {"threshold": threshold, "items": [{"entity_type": r["entity_type"], "similarity": round(float(r["similarity"]), 3),
429+ "a": {"id": r["a_id"], "slug": r["a_slug"], "name": r["a_name"], "organization": r["a_org"], "first_seen_at": r["a_first_seen_at"], "claims": int(r["a_claims"])},
430+ "b": {"id": r["b_id"], "slug": r["b_slug"], "name": r["b_name"], "organization": r["b_org"], "first_seen_at": r["b_first_seen_at"], "claims": int(r["b_claims"])}}
431+ for r in rows]}
432+
433+
434+class MergeBody(BaseModel):
435+ source_id: str
436+ target_id: str
437+
438+
439+@router.post("/entities/merge")
440+async def merge(body: MergeBody) -> dict[str, Any]:
441+ async with transaction() as conn:
442+ src = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=body.source_id)
443+ dst = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=body.target_id)
444+ if not src or not dst:
445+ raise ApiError(404, "source or target entity not found")
446+ try:
447+ result = await merge_entities(conn, src["id"], dst["id"])
448+ except (ValueError, LookupError) as exc:
449+ raise ApiError(400, str(exc)) from exc
450+ await cache.cache_invalidate()
451+ return result
452+
453+
454+@router.post("/entities/{entity_id}/claims/{claim_id}/retract")
455+async def retract_claim(entity_id: str, claim_id: str) -> dict[str, Any]:
456+ async with transaction() as conn:
457+ ent = await fetch_one(conn, "select id from entities where id = :k or slug = :k limit 1", k=entity_id)
458+ if not ent:
459+ raise ApiError(404, "entity not found")
460+ c = await fetch_one(conn, "select * from claims where id = :c and entity_id = :e", c=claim_id, e=ent["id"])
461+ if not c:
462+ raise ApiError(404, "claim not found for this entity")
463+ was_current = c["status"] == "current"
464+ await execute(conn, "update claims set status = 'retracted', valid_to = coalesce(valid_to, now()) where id = :c", c=claim_id)
465+ restored = None
466+ if was_current:
467+ prev = await fetch_one(conn, """select * from claims where entity_id = :e and property = :p and id <> :c and status in ('superseded', 'conflicting')
468+ order by tier, valid_from desc limit 1""", e=ent["id"], p=c["property"], c=claim_id)
469+ if prev:
470+ await execute(conn, "update claims set status = 'current', valid_to = null where id = :id", id=prev["id"])
471+ prov = {"source_id": prev["source_id"], "snapshot_id": prev["snapshot_id"], "url": prev["source_url"], "tier": prev["tier"], "confidence": prev["confidence"],
472+ "extractor": prev["extractor"], "observed_at": prev["observed_at"].isoformat() if prev["observed_at"] else None, "restored_by": "retraction"}
473+ await execute(conn, "update entities set attributes = attributes || jsonb_build_object(:p, cast(:v as jsonb)), provenance = provenance || jsonb_build_object(:p, cast(:pv as jsonb)), updated_at = now() where id = :e",
474+ p=c["property"], v=jsonb(prev["value"]), pv=jsonb(prov), e=ent["id"])
475+ restored = prev["id"]
476+ else:
477+ await execute(conn, "update entities set attributes = attributes - :p, provenance = provenance - :p, updated_at = now() where id = :e", p=c["property"], e=ent["id"])
478+ await execute(conn, """insert into change_events (id, entity_id, event_type, category, property, old_value, summary, importance, connector_name, dedupe_key)
479+ values (:id, :e, 'CLAIM_RETRACTED', 'source', :p, cast(:v as jsonb), :s, 1, 'curation', :dk) on conflict (dedupe_key) do nothing""",
480+ id=new_id("change_event"), e=ent["id"], p=c["property"], v=jsonb(c["value"]),
481+ s=f"Retracted claim {c['property']}", dk=f"retract:{claim_id}")
482+ await cache.cache_invalidate()
483+ return {"ok": True, "claim_id": claim_id, "property": c["property"], "was_current": was_current, "restored_claim_id": restored}
484+
485+
486+# ------------------------------------------------------------------------------------------------------------------ infrastructure & maintenance
487+
488+
489+@router.get("/infrastructure")
490+async def infrastructure() -> dict[str, Any]:
491+ async with connection() as conn:
492+ db = await fetch_one(conn, """select pg_database_size(current_database()) as db_bytes, current_database() as database, version() as pg_version,
493+ (select count(*) from pg_stat_activity where datname = current_database()) as connections,
494+ (select setting from pg_settings where name = 'max_connections') as max_connections""")
495+ tables = await fetch_all(conn, """select c.relname as table, pg_total_relation_size(c.oid) as total_bytes, pg_relation_size(c.oid) as data_bytes,
496+ coalesce(s.n_live_tup, 0) as rows_estimate from pg_class c join pg_namespace n on n.oid = c.relnamespace
497+ left join pg_stat_user_tables s on s.relid = c.oid where n.nspname = 'public' and c.relkind = 'r'
498+ order by pg_total_relation_size(c.oid) desc limit 20""")
499+ extensions = await fetch_all(conn, "select extname, extversion from pg_extension order by 1")
500+ redis_info: dict[str, Any] = {"ok": False}
501+ try:
502+ info = await cache.redis().info("memory")
503+ redis_info = {"ok": True, "used_memory": info.get("used_memory"), "used_memory_human": info.get("used_memory_human"),
504+ "api_cache_keys": sum([1 async for _ in cache.redis().scan_iter(match="aia:api:*", count=1000)])}
505+ except Exception as exc: # noqa: BLE001
506+ redis_info = {"ok": False, "error": exc.__class__.__name__}
507+ return {"hostname": socket.gethostname(), "python": platform.python_version(), "platform": platform.platform(), "api_uptime_s": int(time.time() - _STARTED),
508+ "env": settings.app_env, "heartbeats": await cache.heartbeats(), "archive": archive.archive_size(), "data_dir_exists": settings.data_dir.exists(),
509+ "database": {**(db or {}), "db_bytes": int((db or {}).get("db_bytes") or 0), "tables": tables, "extensions": extensions}, "redis": redis_info,
510+ "llm": {"available": gateway.available, "engine": gateway.engine.name}, "scheduler_tick_s": settings.scheduler_tick_s}
511+
512+
513+@router.post("/cache/flush")
514+async def cache_flush(prefix: str = "") -> dict[str, Any]:
515+ return {"flushed": await cache.cache_invalidate(prefix)}
516+
517+
518+@router.post("/stats/recompute")
519+async def stats_recompute() -> dict[str, Any]:
520+ from aiatlas.services.stats import compute_stats
521+
522+ counts = await compute_stats()
523+ await cache.cache_invalidate("/api/v1/stats")
524+ return {"ok": True, "entities_total": counts.get("entities_total")}
525+
526+
527+class QualityBody(BaseModel):
528+ entity_ids: list[str] | None = None
529+ limit: int = Field(20000, ge=1, le=200000)
530+
531+
532+@router.post("/quality/recompute")
533+async def quality_recompute(body: QualityBody | None = None) -> dict[str, Any]:
534+ from aiatlas.services.quality import recompute
535+
536+ body = body or QualityBody()
537+ res = await recompute(entity_ids=body.entity_ids, limit=body.limit)
538+ await cache.cache_invalidate()
539+ return {"ok": True, **res}
added src/aiatlas/api/routers/benchmarks.py +83 −0
@@ -0,0 +1,83 @@
1+"""/benchmarks listing, /benchmarks/{slug}, /benchmarks/{slug}/results (leaderboard), /benchmarks/{slug}/history."""
2+from __future__ import annotations
3+
4+from typing import Any
5+
6+from fastapi import APIRouter, Query, Request
7+
8+from aiatlas.api.common import (
9+ ENTITY_COLS,
10+ ENTITY_FROM,
11+ PAGINATION,
12+ RESULT_COLS,
13+ RESULT_FROM,
14+ Pagination,
15+ cached,
16+ entity_cols,
17+ entity_summary,
18+ page,
19+ resolve_entity,
20+ resolve_id,
21+ result_row,
22+)
23+from aiatlas.api.detail import leaderboard
24+from aiatlas.api.routers.entities import detail_for_type
25+from aiatlas.db import connection, fetch_all, fetch_val
26+
27+router = APIRouter(prefix="/api/v1/benchmarks", tags=["benchmarks"])
28+
29+
30+@router.get("")
31+@cached(300)
32+async def list_benchmarks(request: Request, category: str | None = None) -> dict[str, Any]:
33+ where = "e.entity_type = 'benchmark' and e.merged_into is null" + (" and e.attributes->>'category' ilike :cat" if category else "")
34+ async with connection() as conn:
35+ rows = await fetch_all(conn, f"""
36+ select {ENTITY_COLS},
37+ (select count(*) from benchmark_results r where r.benchmark_id = e.id and r.valid_to is null) as result_count,
38+ (select count(distinct r.model_id) from benchmark_results r where r.benchmark_id = e.id and r.valid_to is null) as model_count,
39+ top.score as top_score, {entity_cols("tm", "t_")}
40+ from {ENTITY_FROM}
41+ left join lateral (select r.model_id, r.score from benchmark_results r where r.benchmark_id = e.id and r.valid_to is null
42+ order by case when r.higher_is_better then -r.score else r.score end limit 1) top on true
43+ left join entities tm on tm.id = top.model_id left join entities tmo on tmo.id = tm.organization_id
44+ where {where} order by result_count desc, e.canonical_name""", cat=category)
45+ items = []
46+ for r in rows:
47+ top_model = entity_summary(r, "t_")
48+ items.append({**(entity_summary(r) or {}), "result_count": int(r["result_count"] or 0), "model_count": int(r["model_count"] or 0),
49+ "top": {"model": top_model, "score": r["top_score"]} if top_model else None})
50+ return {"items": items}
51+
52+
53+@router.get("/{slug}")
54+@cached(300)
55+async def get_benchmark(request: Request, slug: str) -> dict[str, Any]:
56+ return await detail_for_type(slug, ("benchmark",))
57+
58+
59+@router.get("/{slug}/results")
60+@cached(300)
61+async def benchmark_results(request: Request, slug: str, p: Pagination = PAGINATION, config: str | None = Query(None, max_length=200), history: int = Query(0, ge=0, le=1)) -> dict[str, Any]:
62+ async with connection() as conn:
63+ bench = await resolve_entity(conn, slug, ("benchmark",))
64+ items = await leaderboard(conn, bench["id"], limit=p.limit, offset=p.offset, config=config, history=bool(history))
65+ where = "r.benchmark_id = :id" + ("" if history else " and r.valid_to is null") + (" and r.config::text ilike :cfg" if config else "")
66+ total = await fetch_val(conn, f"select count(*) from benchmark_results r where {where}", id=bench["id"], cfg=f"%{config}%" if config else None)
67+ out = page(items, int(total or 0), p)
68+ out["benchmark"] = entity_summary(bench)
69+ return out
70+
71+
72+@router.get("/{slug}/history")
73+@cached(300)
74+async def benchmark_history(request: Request, slug: str, model: str | None = None, limit: int = Query(1000, ge=1, le=5000)) -> dict[str, Any]:
75+ async with connection() as conn:
76+ bench = await resolve_entity(conn, slug, ("benchmark",))
77+ where = ["r.benchmark_id = :id"]
78+ params: dict[str, Any] = {"id": bench["id"], "lim": limit}
79+ if model:
80+ where.append("r.model_id = :model")
81+ params["model"] = await resolve_id(conn, model)
82+ rows = await fetch_all(conn, f"select {RESULT_COLS} from {RESULT_FROM} where {' and '.join(where)} order by r.observed_at asc, r.id limit :lim", **params)
83+ return {"benchmark": entity_summary(bench), "items": [result_row(r) for r in rows]}
added src/aiatlas/api/routers/changes.py +125 −0
Large diff (125 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/companies.py +64 −0
Large diff (64 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/compare.py +116 −0
Large diff (116 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/diff.py +85 −0
Large diff (85 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/entities.py +152 −0
Large diff (152 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/explore.py +84 −0
Large diff (84 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/hardware.py +92 −0
Large diff (92 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/misc.py +101 −0
Large diff (101 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/models.py +143 −0
Large diff (143 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/papers.py +60 −0
Large diff (60 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/prices.py +90 −0
Large diff (90 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/providers.py +35 −0
Large diff (35 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/search.py +111 −0
Large diff (111 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/sources.py +34 −0
Large diff (34 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/stats.py +33 −0
Large diff (33 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/api/routers/timeline.py +43 −0
Large diff (43 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/services/hardware_fit.py +68 −0
Large diff (68 lines) is not loaded by default to keep the page fast.Load diff
added src/aiatlas/services/merge.py +77 −0
Large diff (77 lines) is not loaded by default to keep the page fast.Load diff
modified src/aiatlas/services/scheduler.py +21 −4
Large diff (41 lines) is not loaded by default to keep the page fast.Load diff
added tests/test_api.py +258 −0
Large diff (259 lines) is not loaded by default to keep the page fast.Load diff
Menu

Theme
↑↓ navigate ↵ open esc close / or ⌘K anywhere