"""/entities/{slug_or_id} and its sub-resources (timeline, history, asof, graph, sources, related, claims, provenance).""" from __future__ import annotations from datetime import UTC, datetime from datetime import time as dtime from typing import Any from fastapi import APIRouter, Query, Request from aiatlas.api.common import ( CLAIM_COLS, CLAIM_FROM, ENTITY_COLS, ENTITY_FROM, ApiError, cached, claim_row, entity_summary, parse_date, parse_ts, resolve_entity, ) from aiatlas.api.detail import entity_detail, sources_of, timeline_of from aiatlas.db import connection, fetch_all, fetch_one, fetch_val router = APIRouter(prefix="/api/v1/entities", tags=["entities"]) @router.get("/{slug_or_id}") @cached(300) async def get_entity(request: Request, slug_or_id: str) -> dict[str, Any]: async with connection() as conn: row = await resolve_entity(conn, slug_or_id) return await entity_detail(row) @router.get("/{slug_or_id}/timeline") @cached(60) async def entity_timeline(request: Request, slug_or_id: str, limit: int = Query(50, ge=1, le=200), before: str | None = None, include_documents: int = Query(0, ge=0, le=1), include_backfill: int = Query(0, ge=0, le=1), date_field: str = Query("occurred", pattern="^(occurred|observed)$")) -> dict[str, Any]: before_ts = parse_ts(before, "before") async with connection() as conn: row = await resolve_entity(conn, slug_or_id) items = await timeline_of(conn, row["id"], row["entity_type"], limit=limit, before=before_ts, include_documents=bool(include_documents), include_backfill=bool(include_backfill), date_field=date_field) cursor_key = "observed_at" if date_field == "observed" else "occurred_at" return {"items": items, "next_before": items[-1][cursor_key] if len(items) == limit else None, "date_field": date_field, "include_backfill": bool(include_backfill)} @router.get("/{slug_or_id}/history") @cached(120) async def entity_history(request: Request, slug_or_id: str, property: str | None = Query(None, max_length=120), limit: int = Query(500, ge=1, le=2000)) -> dict[str, Any]: async with connection() as conn: row = await resolve_entity(conn, slug_or_id) where = "c.entity_id = :id" + (" and c.property = :p" if property else "") rows = await fetch_all(conn, f"select {CLAIM_COLS} from {CLAIM_FROM} where {where} order by c.valid_from desc, c.observed_at desc limit :lim", id=row["id"], p=property, lim=limit) return {"items": [claim_row(r) for r in rows]} @router.get("/{slug_or_id}/claims") @cached(120) async def entity_claims(request: Request, slug_or_id: str, property: str | None = Query(None, max_length=120), status: str | None = Query(None, max_length=40), limit: int = Query(200, ge=1, le=2000), offset: int = Query(0, ge=0)) -> dict[str, Any]: """Public claim list (current by default; `status=all|superseded|conflicting|retracted`), newest first, with claim ids for `/claims/{id}`.""" async with connection() as conn: row = await resolve_entity(conn, slug_or_id) where = ["c.entity_id = :id"] params: dict[str, Any] = {"id": row["id"], "lim": limit, "off": offset} if property: where.append("c.property = :p") params["p"] = property st = status or "current" if st != "all": where.append("c.status = :st") params["st"] = st rows = await fetch_all(conn, f"select {CLAIM_COLS}, c.entity_id, c.snapshot_id, c.run_id, c.extractor_version, c.value_raw from {CLAIM_FROM} where {' and '.join(where)} " f"order by c.property, c.valid_from desc limit :lim offset :off", **params) total = await fetch_val(conn, f"select count(*) from claims c where {' and '.join(where)}", **{k: v for k, v in params.items() if k not in ('lim', 'off')}) items = [{**claim_row(r), "snapshot_id": r.get("snapshot_id"), "run_id": r.get("run_id"), "value_raw": r.get("value_raw")} for r in rows] return {"entity": entity_summary(row), "items": items, "total": int(total or 0), "limit": limit, "offset": offset, "status": st} @router.get("/{slug_or_id}/provenance/{property}") @cached(120) async def entity_provenance(request: Request, slug_or_id: str, property: str) -> dict[str, Any]: """Evidence-drawer payload for one property: current value, source, tier, extractor, confidence, conflicts, history count, snapshot id (no raw content).""" async with connection() as conn: row = await resolve_entity(conn, slug_or_id) cur = await fetch_one(conn, f"select {CLAIM_COLS}, c.snapshot_id, c.run_id, c.extractor_version, c.value_raw, s.tier as source_tier, s.domain as source_domain, " f"s.id as source_id from {CLAIM_FROM} where c.entity_id = :id and c.property = :p and c.status = 'current' order by c.tier, c.valid_from desc limit 1", id=row["id"], p=property) conflicts = await fetch_all(conn, f"select {CLAIM_COLS}, c.snapshot_id from {CLAIM_FROM} where c.entity_id = :id and c.property = :p and c.status = 'conflicting' " f"order by c.tier, c.observed_at desc limit 20", id=row["id"], p=property) history = await fetch_val(conn, "select count(*) from claims c where c.entity_id = :id and c.property = :p", id=row["id"], p=property) snap = None if cur and cur.get("snapshot_id"): snap = await fetch_one(conn, "select s.id, s.observed_at, d.url as document_url, d.title, s.raw_path is not null as archived from snapshots s join documents d on d.id = s.document_id where s.id = :id", id=cur["snapshot_id"]) attrs = row.get("attributes") or {} prov = (row.get("provenance") or {}).get(property) or {} if not cur and property not in attrs: raise ApiError(404, f"no claim for property {property!r}") return {"entity": {"id": row["id"], "slug": row["slug"], "name": row["canonical_name"], "entity_type": row["entity_type"]}, "property": property, "value": cur["value"] if cur else attrs.get(property), "value_raw": cur.get("value_raw") if cur else None, "unit": (cur or {}).get("unit") or prov.get("unit"), "source": {"id": (cur or {}).get("source_id") or prov.get("source_id"), "name": (cur or {}).get("source_name") or prov.get("source_name"), "domain": (cur or {}).get("source_domain"), "url": (cur or {}).get("source_url") or prov.get("url")}, "tier": (cur or {}).get("tier") or prov.get("tier"), "confidence": (cur or {}).get("confidence") or prov.get("confidence"), "extractor": (cur or {}).get("extractor") or prov.get("extractor"), "extractor_version": (cur or {}).get("extractor_version"), "observed_at": (cur or {}).get("observed_at") or prov.get("observed_at"), "effective_at": (cur or {}).get("effective_at"), "valid_since": (cur or {}).get("valid_from"), "claim_id": (cur or {}).get("id"), "run_id": (cur or {}).get("run_id"), "snapshot_id": (cur or {}).get("snapshot_id"), "snapshot": snap, "conflicts": [claim_row(c) for c in conflicts], "history_count": int(history or 0), "note": None if cur else "value materialised in attributes without a current claim row (curated or inherited)"} @router.get("/{slug_or_id}/asof") @cached(300) async def entity_asof(request: Request, slug_or_id: str, date: str = Query(..., description="YYYY-MM-DD")) -> dict[str, Any]: d = parse_date(date, "date") assert d is not None at = datetime.combine(d, dtime.max, UTC) # end of that day, UTC async with connection() as conn: row = await resolve_entity(conn, slug_or_id) existed = row["first_seen_at"] is not None and row["first_seen_at"] <= at rows = await fetch_all(conn, f"""select distinct on (c.property) {CLAIM_COLS} from {CLAIM_FROM} where c.entity_id = :id and c.status <> 'retracted' and c.valid_from <= :at and (c.valid_to is null or c.valid_to > :at) order by c.property, c.tier, c.valid_from desc""", id=row["id"], at=at) if existed else [] claims = [claim_row(r) for r in rows] return {"id": row["id"], "slug": row["slug"], "name": row["canonical_name"], "entity_type": row["entity_type"], "existed": bool(existed), "first_seen_at": row["first_seen_at"], "date": d.isoformat(), "attributes": {c["property"]: c["value"] for c in claims}, "claims": claims} @router.get("/{slug_or_id}/graph") @cached(300) async def entity_graph(request: Request, slug_or_id: str, depth: int = Query(1, ge=1, le=2), limit: int = Query(80, ge=2, le=300)) -> dict[str, Any]: async with connection() as conn: row = await resolve_entity(conn, slug_or_id) root = row["id"] seen = {root} frontier = [root] edges: list[dict[str, str]] = [] for _ in range(depth): if not frontier or len(seen) >= limit: break rows = await fetch_all(conn, """select r.subject_id, r.predicate, r.object_id from relations r where r.valid_to is null and (r.subject_id = any(cast(:ids as text[])) or r.object_id = any(cast(:ids as text[]))) order by r.observed_at desc limit :lim""", ids=frontier, lim=limit * 3) nxt: list[str] = [] for r in rows: other = r["object_id"] if r["subject_id"] in seen else r["subject_id"] if other not in seen: if len(seen) >= limit: continue seen.add(other) nxt.append(other) edges.append({"source": r["subject_id"], "target": r["object_id"], "predicate": r["predicate"]}) frontier = nxt nodes = await fetch_all(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.id = any(cast(:ids as text[]))", ids=list(seen)) node_ids = {n["id"] for n in nodes} uniq = {(e["source"], e["target"], e["predicate"]): e for e in edges if e["source"] in node_ids and e["target"] in node_ids} return {"root": root, "nodes": [{"id": n["id"], "slug": n["slug"], "name": n["canonical_name"], "entity_type": n["entity_type"], "organization_name": n["organization_name"]} for n in nodes], "edges": list(uniq.values())} @router.get("/{slug_or_id}/sources") @cached(300) async def entity_sources(request: Request, slug_or_id: str) -> dict[str, Any]: async with connection() as conn: row = await resolve_entity(conn, slug_or_id) items = await sources_of(conn, row["id"], limit=200) return {"items": items} @router.get("/{slug_or_id}/related") @cached(300) async def entity_related(request: Request, slug_or_id: str, limit: int = Query(12, ge=1, le=60)) -> dict[str, Any]: async with connection() as conn: row = await resolve_entity(conn, slug_or_id) fam = (row.get("attributes") or {}).get("family") rows = await fetch_all(conn, f""" with cand as ( select e.id, 3 as w from entities e where e.entity_type = :t and e.organization_id is not null and e.organization_id = :org and e.id <> :id union all select e.id, 4 from entities e where :fam <> '' and e.entity_type = :t and e.attributes->>'family' = :fam and e.id <> :id union all select e.id, 5 from entities e where :fid <> '' and e.entity_type = :t and e.family_id = :fid and e.id <> :id union all select r2.subject_id, 2 from relations r1 join relations r2 on r2.object_id = r1.object_id and r2.predicate = r1.predicate where r1.subject_id = :id and r1.valid_to is null and r2.valid_to is null and r2.subject_id <> :id union all select r2.object_id, 1 from relations r1 join relations r2 on r2.subject_id = r1.subject_id and r2.predicate = r1.predicate where r1.object_id = :id and r1.valid_to is null and r2.valid_to is null and r2.object_id <> :id), scored as (select id, sum(w) as w from cand group by id) select {ENTITY_COLS} from scored join entities e on e.id = scored.id left join entities eo on eo.id = e.organization_id where e.merged_into is null order by scored.w desc, coalesce((e.quality->>'score')::float, 0) desc, e.updated_at desc limit :lim""", id=row["id"], t=row["entity_type"], org=row.get("organization_id") or "", fam=str(fam or ""), fid=row.get("family_id") or "", lim=limit) return {"items": [entity_summary(r) for r in rows]} async def detail_for_type(slug: str, types: tuple[str, ...]) -> dict[str, Any]: async with connection() as conn: row = await resolve_entity(conn, slug, types) return await entity_detail(row) async def entity_exists(slug: str) -> dict[str, Any] | None: async with connection() as conn: return await fetch_one(conn, "select id, entity_type from entities where slug = :s or id = :s limit 1", s=slug) __all__ = ["ApiError", "detail_for_type", "router"]