"""`EntityDetail` builder (docs/API.md): shared by `/entities/{slug}` and the type-scoped aliases. API 1.1: blocks run in at most FOUR concurrent groups (one pooled connection each, sequential inside a group) instead of one connection per block; model details gain `family`, `artifacts`, `deployments`, `identity`, `licence`, `openness`, `version_history` and grouped `benchmarks`; artifact details carry `canonical` + `artifact_kind`; a resolved `merged_into` hop is reported as `redirected_from`.""" from __future__ import annotations import asyncio from collections import defaultdict from typing import Any from sqlalchemy.ext.asyncio import AsyncConnection from aiatlas.api.common import ( ARTIFACT_KINDS, CLAIM_COLS, CLAIM_FROM, COMPANY_TYPES, ENTITY_COLS, ENTITY_FROM, EVENT_COLS, EVENT_FROM, PRICE_COLS, PRICE_FROM, RESULT_COLS, RESULT_FROM, RESULT_ORDER, change_event, deployment_row, enrich_provenance, entity_summary, price_row, result_row, ) from aiatlas.db import connection, fetch_all, fetch_one from aiatlas.ontology.benchmarks import TRUST_LABELS from aiatlas.ontology.licenses import LICENSES, normalize_license from aiatlas.ontology.openness import OPENNESS_DEFINITIONS, OPENNESS_LABELS, normalize_openness, openness_dimensions from aiatlas.services import hardware_fit as hf from aiatlas.services.frontier import config_summary, enrich, group_label LINEAGE_PREDICATES = ("derived_from", "fine_tuned_from", "distilled_from", "merged_from", "quantized_from") RELATION_GROUP_LIMIT = 24 VERSIONED_PROPERTIES = ("context_length", "max_output_tokens", "status", "knowledge_cutoff", "license", "openness", "parameter_count") MAX_CONCURRENT_GROUPS = 4 async def relations_grouped(conn: AsyncConnection, entity_id: str) -> list[dict[str, Any]]: rows = await fetch_all(conn, f""" with rel as ( 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 union all 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), ranked as (select rel.*, row_number() over (partition by predicate, direction order by observed_at desc) as rn, count(*) over (partition by predicate, direction) as total from rel) 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 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) groups: dict[tuple[str, str], dict[str, Any]] = {} for r in rows: g = groups.setdefault((r["predicate"], r["direction"]), {"predicate": r["predicate"], "direction": r["direction"], "items": [], "total": int(r["total"])}) g["items"].append(entity_summary(r)) return list(groups.values()) async def sources_of(conn: AsyncConnection, entity_id: str, limit: int = 60) -> list[dict[str, Any]]: """Documents attached to the entity + documents whose snapshots back its claims (deduped by URL — `documents.url` is unique).""" rows = await fetch_all(conn, """ with docs as (select d.id from documents d where d.entity_id = :id union select s.document_id from claims c join snapshots s on s.id = c.snapshot_id where c.entity_id = :id 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) 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, s.domain, s.tier, (select count(*) from snapshots x where x.document_id = d.id) as snapshots from docs join documents d on d.id = docs.id left join sources s on s.id = d.source_id order by s.tier nulls last, last_observed_at desc nulls last limit :lim""", id=entity_id, lim=limit) 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"], "title": r.get("title"), "last_observed_at": r["last_observed_at"], "snapshots": int(r["snapshots"] or 0)} for r in rows] def _scope_sql(is_org: bool) -> str: if is_org: return ("(ev.entity_id = :id or ev.entity_id in (select id from entities where organization_id = :id union " "select object_id from relations where subject_id = :id and predicate in ('develops','owns','operates','published') and valid_to is null))") return "ev.entity_id = :id" def event_date_col(date_field: str) -> str: return "ev.observed_at" if date_field == "observed" else "ev.occurred_at" async def timeline_of(conn: AsyncConnection, entity_id: str, entity_type: str, *, limit: int = 30, before: Any = None, include_documents: bool = False, include_backfill: bool = False, date_field: str = "occurred") -> list[dict[str, Any]]: col = event_date_col(date_field) where = [_scope_sql(entity_type in COMPANY_TYPES)] params: dict[str, Any] = {"id": entity_id, "lim": limit} if before is not None: where.append(f"{col} < :before") params["before"] = before if not include_documents: where.append("ev.event_type <> 'DOCUMENT_CHANGED'") if not include_backfill: where.append("ev.is_backfill = false") rows = await fetch_all(conn, f"select {EVENT_COLS}, ev.occurred_at, ev.is_backfill, ev.group_key from {EVENT_FROM} where {' and '.join(where)} " f"order by {col} desc, ev.id desc limit :lim", **params) out = [] for r in rows: ev = change_event(r) ev["occurred_at"], ev["is_backfill"], ev["group_key"] = r.get("occurred_at"), r.get("is_backfill"), r.get("group_key") out.append(ev) return out async def prices_of_model(conn: AsyncConnection, model_id: str, *, current_only: bool) -> list[dict[str, Any]]: cond = "and p.valid_to is null" if current_only else "" order = "p.input_per_mtok nulls last, pv.canonical_name" if current_only else "p.valid_from, p.id" 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) return [price_row(r) for r in rows] async def deployments_of_model(conn: AsyncConnection, model_id: str) -> list[dict[str, Any]]: rows = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where p.model_id = :id and p.valid_to is null order by p.output_per_mtok nulls last, pv.canonical_name limit 200", id=model_id) return [deployment_row(r) for r in rows] async def prices_of_provider(conn: AsyncConnection, provider_id: str) -> list[dict[str, Any]]: 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) return [price_row(r) for r in rows] async def deployments_of_provider(conn: AsyncConnection, provider_id: str) -> list[dict[str, Any]]: """Current offers of this provider, cheapest output first.""" 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 p.output_per_mtok asc nulls last, m.canonical_name limit 500", id=provider_id) return [deployment_row(r) for r in rows] async def removed_deployments_of_provider(conn: AsyncConnection, provider_id: str, *, days: int = 90) -> list[dict[str, Any]]: """Offers this provider closed in the last `days` days (status delisted).""" rows = await fetch_all(conn, f"select {PRICE_COLS} from {PRICE_FROM} where p.provider_id = :id and p.valid_to > now() - make_interval(days => :d) order by p.valid_to desc limit 200", id=provider_id, d=days) return [deployment_row(r) for r in rows] async def results_of_model(conn: AsyncConnection, model_id: str) -> list[dict[str, Any]]: 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) return [result_row(r) for r in rows] async def benchmarks_of_model(conn: AsyncConnection, model_id: str) -> dict[str, Any]: """Current results grouped by benchmark → metric → (config group): best row, n_rows, trust level, comparability group label.""" rows = await fetch_all(conn, """ select r.id, r.model_id, r.benchmark_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.config_key, r.trust_level, r.extractor, s.key as source_key, b.slug as benchmark_slug, b.canonical_name as benchmark_name, b.attributes->>'category' as category from benchmark_results r join entities b on b.id = r.benchmark_id left join sources s on s.id = r.source_id where r.model_id = :id and r.valid_to is null and r.is_current order by b.canonical_name, r.observed_at desc limit 500""", id=model_id) for r in rows: r["model_name"] = None enrich(r) by_bench: dict[str, dict[str, Any]] = {} for r in rows: b = by_bench.setdefault(r["benchmark_id"], {"benchmark": {"id": r["benchmark_id"], "slug": r["benchmark_slug"], "name": r["benchmark_name"], "category": r["category"]}, "metrics": {}}) m = b["metrics"].setdefault(r["metric_canonical"], {"metric": r["metric_canonical"], "groups": {}}) g = m["groups"].setdefault(r["config_key"], {"config_key": r["config_key"], "label": group_label(r["metric_canonical"], r.get("config")), "rows": []}) g["rows"].append(r) items = [] for b in by_bench.values(): metrics = [] for m in b["metrics"].values(): groups = [] for g in m["groups"].values(): hib = all(x.get("higher_is_better", True) for x in g["rows"]) best = max(g["rows"], key=lambda x: x["score"]) if hib else min(g["rows"], key=lambda x: x["score"]) groups.append({"config_key": g["config_key"], "comparability_group": g["label"], "n_rows": len(g["rows"]), "higher_is_better": hib, "best": {"score": best["score"], "unit": best.get("unit"), "trust_level": best["trust_level"], "trust_label": TRUST_LABELS.get(best["trust_level"], best["trust_level"]), "config": config_summary(best.get("config")), "evaluated_at": best.get("evaluated_at"), "observed_at": best["observed_at"], "source_url": best.get("source_url"), "tier": best.get("tier"), "result_id": best["id"]}, "trust_levels": sorted({x["trust_level"] for x in g["rows"]})}) metrics.append({"metric": m["metric"], "groups": groups}) items.append({**b["benchmark"], "metrics": metrics}) return {"items": items, "total_rows": len(rows), "note": "Current rows only, grouped by benchmark → canonical metric → comparability group (task configuration). " "Effort variants folded into this model appear as rows of the same group."} async def leaderboard(conn: AsyncConnection, benchmark_id: str, *, limit: int = 100, offset: int = 0, config: str | None = None, history: bool = False, metric: str | None = None, config_key: str | None = None) -> list[dict[str, Any]]: where = ["r.benchmark_id = :id"] params: dict[str, Any] = {"id": benchmark_id, "lim": limit, "off": offset} if not history: where.append("r.valid_to is null") if config: where.append("r.config::text ilike :cfg") params["cfg"] = f"%{config}%" if metric: where.append("lower(r.metric) = lower(:metric)") params["metric"] = metric if config_key: where.append("r.config_key = :ck") params["ck"] = config_key rows = await fetch_all(conn, f"select {RESULT_COLS}, r.config_key, r.trust_level from {RESULT_FROM} where {' and '.join(where)} order by {RESULT_ORDER} limit :lim offset :off", **params) out = [] for r in rows: item = result_row(r) item["config_key"] = r.get("config_key") item["trust_level"] = r.get("trust_level") out.append(item) return out async def lineage_of(conn: AsyncConnection, model_id: str) -> dict[str, list[dict[str, Any]]]: preds = list(LINEAGE_PREDICATES) desc_preds = [p for p in preds if p != "quantized_from"] ancestors = await fetch_all(conn, f""" with recursive up as ( 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[])) union 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) 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 where e.id <> :id and e.merged_into is null order by e.id, up.depth""", id=model_id, preds=preds) descendants = await fetch_all(conn, f""" with recursive down as ( 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[])) union 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) 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 where e.id <> :id and e.merged_into is null and e.entity_type <> 'artifact' order by e.id, down.depth""", id=model_id, preds=desc_preds) 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 where r.object_id = :id and r.predicate = 'quantized_from' and r.valid_to is null and e.merged_into is null order by e.canonical_name limit 100""", id=model_id) key = lambda r: (r.get("depth", 0), r["canonical_name"] or "") return {"ancestors": [entity_summary(r) for r in sorted(ancestors, key=key)], "descendants": [entity_summary(r) for r in sorted(descendants, key=key)], "quantizations": [entity_summary(r) for r in quants]} async def artifacts_of_model(conn: AsyncConnection, model_id: str) -> dict[str, Any]: """Artifacts (entity_type 'artifact' with canonical_id = model, or `artifact_of` relation) grouped by kind.""" rows = await fetch_all(conn, f""" with ids as (select e.id, e.artifact_kind from entities e where e.canonical_id = :id and e.entity_type = 'artifact' and e.merged_into is null union select r.subject_id, null from relations r join entities x on x.id = r.subject_id where r.object_id = :id and r.predicate in ('artifact_of','quantized_from') and r.valid_to is null and x.entity_type = 'artifact' and x.merged_into is null) select distinct on (e.id) coalesce(e.artifact_kind, ids.artifact_kind) as kind, {ENTITY_COLS} from ids join entities e on e.id = ids.id left join entities eo on eo.id = e.organization_id order by e.id limit 300""", id=model_id) groups: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in rows: kind = r["kind"] if r["kind"] in ARTIFACT_KINDS else "other" s = entity_summary(r) or {} s["artifact_kind"] = r["kind"] groups[kind].append(s) ordered = [k for k in (*ARTIFACT_KINDS, "other") if k in groups] return {"items": [{"kind": k, "items": sorted(groups[k], key=lambda x: x["name"] or ""), "count": len(groups[k])} for k in ordered], "total": len(rows)} async def family_of_model(conn: AsyncConnection, row: dict[str, Any]) -> dict[str, Any] | None: fid = row.get("family_id") if fid: fam = await fetch_one(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.id = :id", id=fid) if fam: return entity_summary(fam) label = (row.get("attributes") or {}).get("family") if label: fam = await fetch_one(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.entity_type = 'model_family' and e.merged_into is null and e.canonical_name ilike :n limit 1", n=label) return entity_summary(fam) if fam else {"id": None, "entity_type": "model_family", "slug": None, "name": label, "canonical": False, "note": "family label from attributes; no model_family entity yet"} return None async def canonical_of_artifact(conn: AsyncConnection, row: dict[str, Any]) -> dict[str, Any] | None: cid = row.get("canonical_id") if not cid: rel = await fetch_one(conn, "select object_id from relations where subject_id = :id and predicate in ('artifact_of','quantized_from') and valid_to is null order by predicate limit 1", id=row["id"]) cid = rel["object_id"] if rel else None if not cid: return None can = await fetch_one(conn, f"select {ENTITY_COLS} from {ENTITY_FROM} where e.id = :id", id=cid) return entity_summary(can) if can else None async def identity_of_model(conn: AsyncConnection, model_id: str, identifiers: list[dict[str, str]]) -> dict[str, Any]: counts = await fetch_one(conn, """ select (select count(*) from entities a where a.canonical_id = :id and a.entity_type = 'artifact' and a.merged_into is null and coalesce(a.artifact_kind, '') <> 'checkpoint') as third_party, (select count(*) from entities a where a.canonical_id = :id and a.entity_type = 'artifact' and a.merged_into is null and a.artifact_kind = 'checkpoint') as official_artifacts, (select count(distinct p.provider_id) from prices p where p.model_id = :id and p.valid_to is null) as providers, (select count(*) from entities v where v.merged_into = :id) as folded_variants""", id=model_id) hf_repos = [i["value"] for i in identifiers if i["scheme"] == "hf_repo"] api_ids = sorted({i["value"] for i in identifiers if i["scheme"].endswith("_model_id") or i["scheme"] in ("openrouter", "api_model_id", "artificial_analysis")}) c = counts or {} return {"canonical_model": True, "official_checkpoints": hf_repos, "official_artifacts": int(c.get("official_artifacts") or 0), "third_party_artifacts": int(c.get("third_party") or 0), "provider_deployments": int(c.get("providers") or 0), "folded_variants": int(c.get("folded_variants") or 0), "api_aliases": api_ids, "note": "official_checkpoints = hf_repo identifiers carried by the model itself; artifacts are separate entities pointing here through canonical_id."} def licence_block(attrs: dict[str, Any]) -> dict[str, Any] | None: key = attrs.get("license_key") or normalize_license(attrs.get("license")) info = LICENSES.get(key) if key else None if not info: return {"key": None, "raw": attrs.get("license") or attrs.get("license_raw"), "note": "licence label not classified in the ontology"} if attrs.get("license") else None return {**info.as_dict(), "raw": attrs.get("license_raw") or attrs.get("license"), "url_observed": attrs.get("license_url")} def openness_block(attrs: dict[str, Any]) -> dict[str, Any] | None: raw = attrs.get("openness") cat = normalize_openness(raw) if raw else None dims = attrs.get("openness_dimensions") if not isinstance(dims, dict): key = attrs.get("license_key") or normalize_license(attrs.get("license")) weights = True if cat in ("open-weights", "open-source", "restricted-weights") else False if cat == "proprietary" else None dims = openness_dimensions(weights_available=weights, license_key=key, license_raw=attrs.get("license")) if not cat and not raw: return None return {"category": cat or "unknown", "raw": raw, "label": OPENNESS_LABELS.get(cat or "unknown"), "definition": OPENNESS_DEFINITIONS.get(cat or "unknown"), "dimensions": dims, "note": "dimensions marked null are unknown, not false"} async def version_history_of(conn: AsyncConnection, entity_id: str) -> list[dict[str, Any]]: rows = await fetch_all(conn, f"select {CLAIM_COLS} from {CLAIM_FROM} where c.entity_id = :id and c.property = any(cast(:props as text[])) and c.status <> 'retracted' " f"order by c.property, c.valid_from asc, c.observed_at asc limit 2000", id=entity_id, props=list(VERSIONED_PROPERTIES)) by_prop: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in rows: by_prop[r["property"]].append(r) out = [] for prop in VERSIONED_PROPERTIES: claims = by_prop.get(prop) if not claims: continue transitions = [] prev: Any = None for c in claims: if c["status"] == "conflicting": continue if c["value"] == prev and transitions: transitions[-1]["valid_to"] = c["valid_to"] or transitions[-1]["valid_to"] continue transitions.append({"from": prev, "to": c["value"], "valid_from": c["valid_from"], "valid_to": c["valid_to"], "effective_at": c["effective_at"], "source_url": c["source_url"], "tier": c["tier"], "claim_id": c["id"], "status": c["status"]}) prev = c["value"] out.append({"property": prop, "transitions": transitions, "current": transitions[-1]["to"] if transitions else None}) return out async def providers_of_model(conn: AsyncConnection, model_id: str) -> list[dict[str, Any]]: rows = await fetch_all(conn, f""" 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 union select p.provider_id from prices p where p.model_id = :id and p.valid_to is null) select {ENTITY_COLS} from ids join entities e on e.id = ids.id left join entities eo on eo.id = e.organization_id where e.merged_into is null order by e.canonical_name limit 100""", id=model_id) return [entity_summary(r) for r in rows] async def hardware_fit_of_model(conn: AsyncConnection, attrs: dict[str, Any]) -> list[dict[str, Any]] | None: params = hf.parameter_count(attrs) if params is None: return None context = 8192 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") out: list[dict[str, Any]] = [] for r in rows: mem = hf.hardware_memory_gb(r["attributes"]) if mem is None: continue f = hf.fit(params, mem, "4bit", context) out.append({"hardware": entity_summary(r), "memory_gb": mem, **f}) out.sort(key=lambda x: (not x["fits"], -x["headroom_gb"])) return out 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]: """Entities of `etype` linked to `entity_id` by any live relation (either direction) — optionally also those whose organization is `entity_id`.""" org_sql = "union select id from entities where organization_id = :id" if include_org_children else "" rows = await fetch_all(conn, f""" with ids as (select r.object_id as id from relations r where r.subject_id = :id and r.valid_to is null union select r.subject_id from relations r where r.object_id = :id and r.valid_to is null {org_sql}) 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 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 limit :lim offset :off""", id=entity_id, t=etype, lim=limit, off=offset) return [entity_summary(r) for r in rows], int(rows[0]["total"]) if rows else 0 async def _run_group(tasks: list[tuple[str, Any, tuple[Any, ...], dict[str, Any]]]) -> dict[str, Any]: """Run the blocks of one group sequentially on a single pooled connection.""" out: dict[str, Any] = {} async with connection() as conn: for name, fn, args, kw in tasks: out[name] = await fn(conn, *args, **kw) return out async def entity_detail(row: dict[str, Any]) -> dict[str, Any]: """Full detail. Blocks are spread over ≤ 4 concurrent groups (one connection each).""" eid, etype = row["id"], row["entity_type"] provenance = dict(row.get("provenance") or {}) attrs = row.get("attributes") or {} async def base(conn: AsyncConnection) -> dict[str, Any]: aliases = await fetch_all(conn, "select alias from entity_aliases where entity_id = :id order by kind, alias limit 200", id=eid) idents = await fetch_all(conn, "select scheme, value from entity_identifiers where entity_id = :id order by scheme, value limit 200", id=eid) await enrich_provenance(conn, provenance) return {"aliases": [a["alias"] for a in aliases], "identifiers": [{"scheme": i["scheme"], "value": i["value"]} for i in idents]} T = lambda name, fn, *args, **kw: (name, fn, args, kw) groups: list[list[tuple[str, Any, tuple[Any, ...], dict[str, Any]]]] = [ [T("base", base), T("relations", relations_grouped, eid)], [T("sources", sources_of, eid), T("timeline", timeline_of, eid, etype)], ] if etype == "model": groups[0] += [T("prices", prices_of_model, eid, current_only=True), T("deployments", deployments_of_model, eid), T("providers", providers_of_model, eid)] groups[1] += [T("price_history", prices_of_model, eid, current_only=False), T("family", family_of_model, row), T("artifacts", artifacts_of_model, eid)] groups.append([T("results", results_of_model, eid), T("benchmarks", benchmarks_of_model, eid), T("version_history", version_history_of, eid)]) groups.append([T("lineage", lineage_of, eid), T("hardware_fit", hardware_fit_of_model, attrs), T("papers", related_by_type, eid, "paper", limit=24), T("repositories", related_by_type, eid, "repository", limit=24)]) elif etype == "artifact": groups[0] += [T("canonical", canonical_of_artifact, row), T("prices", prices_of_model, eid, current_only=True)] groups[1] += [T("lineage", lineage_of, eid), T("hardware_fit", hardware_fit_of_model, attrs)] elif etype in COMPANY_TYPES: groups[0] += [T("models", related_by_type, eid, "model", limit=50, include_org_children=True)] groups[1] += [T("papers", related_by_type, eid, "paper", limit=24, include_org_children=True), T("repositories", related_by_type, eid, "repository", limit=24, include_org_children=True)] elif etype == "provider": groups[0] += [T("prices", prices_of_provider, eid), T("deployments", deployments_of_provider, eid)] groups[1] += [T("models", _provider_models, eid), T("removed", removed_deployments_of_provider, eid)] elif etype == "benchmark": groups[0] += [T("results", leaderboard, eid, limit=100)] elif etype == "hardware": groups[0] += [T("models", related_by_type, eid, "model", limit=50)] elif etype in ("framework", "library", "runtime"): groups[0] += [T("repositories", related_by_type, eid, "repository", limit=24)] elif etype == "model_family": groups[0] += [T("models", _family_models, eid)] results = await asyncio.gather(*(_run_group(g) for g in groups[:MAX_CONCURRENT_GROUPS])) blocks: dict[str, Any] = {} for r in results: blocks.update(r) detail = entity_summary(row) or {} detail["attributes"] = attrs detail["provenance"] = provenance detail.update(blocks.pop("base")) detail["relations"] = blocks.pop("relations") detail["sources"] = blocks.pop("sources") detail["timeline"] = blocks.pop("timeline") for k, v in blocks.items(): if k in ("models",) and isinstance(v, tuple): items, total = v detail[k] = {"items": items, "total": total, "limit": 50, "offset": 0} elif k in ("papers", "repositories") and isinstance(v, tuple): detail[k] = v[0] elif v is not None: detail[k] = v if etype == "model": if "hardware_fit" in detail: detail["hardware_fit_assumptions"] = hf.ASSUMPTIONS detail["identity"] = await _with_conn(identity_of_model, eid, detail.get("identifiers") or []) lic = licence_block(attrs) if lic is not None: detail["licence"] = lic opn = openness_block(attrs) if opn is not None: detail["openness"] = opn detail["family_id"] = row.get("family_id") detail["identity_confidence"] = row.get("identity_confidence") if etype == "artifact": detail["artifact_kind"] = row.get("artifact_kind") detail.setdefault("canonical", None) if "hardware_fit" in detail: detail["hardware_fit_assumptions"] = hf.ASSUMPTIONS if row.get("redirected_from"): detail["redirected_from"] = row["redirected_from"] return detail async def _with_conn(fn: Any, *args: Any) -> Any: async with connection() as conn: return await fn(conn, *args) async def _provider_models(conn: AsyncConnection, provider_id: str) -> tuple[list[dict[str, Any]], int]: rows = await fetch_all(conn, f""" with ids as (select p.model_id as id from prices p where p.provider_id = :id and p.valid_to is null union select r.subject_id from relations r where r.object_id = :id and r.predicate = 'available_through' and r.valid_to is null) 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 where e.merged_into is null order by e.canonical_name limit 50""", id=provider_id) return [entity_summary(r) for r in rows], int(rows[0]["total"]) if rows else 0 async def _family_models(conn: AsyncConnection, family_id: str) -> tuple[list[dict[str, Any]], int]: rows = await fetch_all(conn, f"""select count(*) over () as total, {ENTITY_COLS} from {ENTITY_FROM} where e.family_id = :id and e.entity_type = 'model' and e.merged_into is null order by e.attributes->>'release_date' desc nulls last, e.canonical_name limit 50""", id=family_id) return [entity_summary(r) for r in rows], int(rows[0]["total"]) if rows else 0 __all__ = [ "VERSIONED_PROPERTIES", "artifacts_of_model", "benchmarks_of_model", "deployments_of_model", "entity_detail", "event_date_col", "family_of_model", "hardware_fit_of_model", "leaderboard", "licence_block", "lineage_of", "openness_block", "prices_of_model", "prices_of_provider", "providers_of_model", "related_by_type", "relations_grouped", "results_of_model", "sources_of", "timeline_of", "version_history_of", ]