generated from coulomb/repo-seed
Add registry feed and library inventory
This commit is contained in:
@@ -71,6 +71,27 @@ class RegistryStore:
|
||||
|
||||
create index if not exists idx_artifacts_target
|
||||
on artifacts(target_id);
|
||||
|
||||
create table if not exists libraries (
|
||||
id integer primary key autoincrement,
|
||||
repo_slug text not null references repositories(slug),
|
||||
bom_ref text,
|
||||
component_type text not null,
|
||||
name text not null,
|
||||
version text,
|
||||
purl text,
|
||||
scope text,
|
||||
licenses_json text not null,
|
||||
hashes_json text not null,
|
||||
metadata_json text not null,
|
||||
created_at text not null
|
||||
);
|
||||
|
||||
create index if not exists idx_libraries_repo
|
||||
on libraries(repo_slug);
|
||||
|
||||
create index if not exists idx_libraries_purl
|
||||
on libraries(purl);
|
||||
"""
|
||||
)
|
||||
|
||||
@@ -316,6 +337,97 @@ class RegistryStore:
|
||||
enriched["artifacts"] = self.list_artifacts(target_id=graph_id)
|
||||
return enriched
|
||||
|
||||
def ingest_cyclonedx(self, repo_slug: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
self.get_repository(repo_slug)
|
||||
bom = payload.get("bom") if "bom" in payload else payload
|
||||
if not isinstance(bom, dict):
|
||||
raise RegistryError("CycloneDX payload must be an object")
|
||||
if bom.get("bomFormat") and bom.get("bomFormat") != "CycloneDX":
|
||||
raise RegistryError("CycloneDX payload must have bomFormat 'CycloneDX'")
|
||||
|
||||
entries = _cyclonedx_entries(bom)
|
||||
now = _utc_now()
|
||||
with self._connect() as db:
|
||||
db.execute("delete from libraries where repo_slug = ?", (repo_slug,))
|
||||
for entry in entries:
|
||||
db.execute(
|
||||
"""
|
||||
insert into libraries (
|
||||
repo_slug, bom_ref, component_type, name, version, purl,
|
||||
scope, licenses_json, hashes_json, metadata_json, created_at
|
||||
)
|
||||
values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
repo_slug,
|
||||
entry["bom_ref"],
|
||||
entry["component_type"],
|
||||
entry["name"],
|
||||
entry["version"],
|
||||
entry["purl"],
|
||||
entry["scope"],
|
||||
json.dumps(entry["licenses"], sort_keys=True),
|
||||
json.dumps(entry["hashes"], sort_keys=True),
|
||||
json.dumps(entry["metadata"], sort_keys=True),
|
||||
now,
|
||||
),
|
||||
)
|
||||
return {
|
||||
"repo_slug": repo_slug,
|
||||
"component_count": len(entries),
|
||||
"libraries": self.list_libraries(repo_slug=repo_slug),
|
||||
}
|
||||
|
||||
def list_libraries(
|
||||
self,
|
||||
repo_slug: str | None = None,
|
||||
name: str | None = None,
|
||||
purl: str | None = None,
|
||||
component_type: str | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
conditions: list[str] = []
|
||||
params: list[str] = []
|
||||
if repo_slug:
|
||||
conditions.append("repo_slug = ?")
|
||||
params.append(repo_slug)
|
||||
if name:
|
||||
conditions.append("name = ?")
|
||||
params.append(name)
|
||||
if purl:
|
||||
conditions.append("purl = ?")
|
||||
params.append(purl)
|
||||
if component_type:
|
||||
conditions.append("component_type = ?")
|
||||
params.append(component_type)
|
||||
where = f" where {' and '.join(conditions)}" if conditions else ""
|
||||
with self._connect() as db:
|
||||
rows = db.execute(
|
||||
"""
|
||||
select id, repo_slug, bom_ref, component_type, name, version, purl,
|
||||
scope, licenses_json, hashes_json, metadata_json, created_at
|
||||
from libraries
|
||||
"""
|
||||
+ where
|
||||
+ " order by repo_slug, name, version, id",
|
||||
params,
|
||||
).fetchall()
|
||||
return [_library_dict(row) for row in rows]
|
||||
|
||||
def get_library(self, library_id: int) -> dict[str, Any]:
|
||||
with self._connect() as db:
|
||||
row = db.execute(
|
||||
"""
|
||||
select id, repo_slug, bom_ref, component_type, name, version, purl,
|
||||
scope, licenses_json, hashes_json, metadata_json, created_at
|
||||
from libraries
|
||||
where id = ?
|
||||
""",
|
||||
(library_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
raise RegistryError(f"library not found: {library_id}", 404)
|
||||
return _library_dict(row)
|
||||
|
||||
def _connect(self) -> sqlite3.Connection:
|
||||
db = sqlite3.connect(self.path)
|
||||
db.row_factory = sqlite3.Row
|
||||
@@ -614,6 +726,30 @@ def xregistry_projection(graph: dict[str, Any]) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def library_xregistry_projection(libraries: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
group = _xregistry_group("library", "libraries")
|
||||
for library in libraries:
|
||||
key = _xregistry_key(str(library.get("purl") or library.get("bom_ref") or library.get("name", "")))
|
||||
group["resources"][key] = {
|
||||
"id": library.get("purl") or library.get("bom_ref") or library.get("name", ""),
|
||||
"name": library.get("name", ""),
|
||||
"versionid": library.get("version") or "unknown",
|
||||
"attributes": {
|
||||
"repo": library.get("repo_slug", ""),
|
||||
"bom_ref": library.get("bom_ref", ""),
|
||||
"component_type": library.get("component_type", ""),
|
||||
"scope": library.get("scope", ""),
|
||||
"licenses": library.get("licenses", []),
|
||||
"hashes": library.get("hashes", []),
|
||||
},
|
||||
}
|
||||
return {
|
||||
"apiVersion": "railiance.fabric/v1alpha1",
|
||||
"kind": "LibraryXRegistryProjection",
|
||||
"groups": {"libraries": group},
|
||||
}
|
||||
|
||||
|
||||
def _with_source(graph: dict[str, Any], repo_slug: str, commit: str, generated_at: str) -> dict[str, Any]:
|
||||
copy = json.loads(json.dumps(graph))
|
||||
copy.setdefault("generated_at", generated_at)
|
||||
@@ -656,6 +792,92 @@ def _artifact_dict(row: sqlite3.Row) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def _library_dict(row: sqlite3.Row) -> dict[str, Any]:
|
||||
return {
|
||||
"id": row["id"],
|
||||
"repo_slug": row["repo_slug"],
|
||||
"bom_ref": row["bom_ref"],
|
||||
"component_type": row["component_type"],
|
||||
"name": row["name"],
|
||||
"version": row["version"],
|
||||
"purl": row["purl"],
|
||||
"scope": row["scope"],
|
||||
"licenses": json.loads(row["licenses_json"]),
|
||||
"hashes": json.loads(row["hashes_json"]),
|
||||
"metadata": json.loads(row["metadata_json"]),
|
||||
"created_at": row["created_at"],
|
||||
}
|
||||
|
||||
|
||||
def _cyclonedx_entries(bom: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
entries: list[dict[str, Any]] = []
|
||||
for component in bom.get("components", []):
|
||||
if not isinstance(component, dict):
|
||||
continue
|
||||
name = str(component.get("name") or "").strip()
|
||||
if not name:
|
||||
continue
|
||||
entries.append(
|
||||
{
|
||||
"bom_ref": _optional_component_text(component, "bom-ref"),
|
||||
"component_type": str(component.get("type") or "library"),
|
||||
"name": name,
|
||||
"version": _optional_component_text(component, "version"),
|
||||
"purl": _optional_component_text(component, "purl"),
|
||||
"scope": _optional_component_text(component, "scope"),
|
||||
"licenses": _normalize_licenses(component.get("licenses", [])),
|
||||
"hashes": component.get("hashes", []) if isinstance(component.get("hashes", []), list) else [],
|
||||
"metadata": {
|
||||
"group": component.get("group"),
|
||||
"publisher": component.get("publisher"),
|
||||
"description": component.get("description"),
|
||||
"externalReferences": component.get("externalReferences", []),
|
||||
},
|
||||
}
|
||||
)
|
||||
for service in bom.get("services", []):
|
||||
if not isinstance(service, dict):
|
||||
continue
|
||||
name = str(service.get("name") or "").strip()
|
||||
if not name:
|
||||
continue
|
||||
entries.append(
|
||||
{
|
||||
"bom_ref": _optional_component_text(service, "bom-ref"),
|
||||
"component_type": "service",
|
||||
"name": name,
|
||||
"version": _optional_component_text(service, "version"),
|
||||
"purl": None,
|
||||
"scope": None,
|
||||
"licenses": [],
|
||||
"hashes": [],
|
||||
"metadata": {
|
||||
"provider": service.get("provider"),
|
||||
"endpoints": service.get("endpoints", []),
|
||||
"externalReferences": service.get("externalReferences", []),
|
||||
},
|
||||
}
|
||||
)
|
||||
return entries
|
||||
|
||||
|
||||
def _optional_component_text(component: dict[str, Any], key: str) -> str | None:
|
||||
value = component.get(key)
|
||||
if value is None:
|
||||
return None
|
||||
return str(value)
|
||||
|
||||
|
||||
def _normalize_licenses(raw: Any) -> list[dict[str, Any]]:
|
||||
if not isinstance(raw, list):
|
||||
return []
|
||||
normalized = []
|
||||
for item in raw:
|
||||
if isinstance(item, dict):
|
||||
normalized.append(item)
|
||||
return normalized
|
||||
|
||||
|
||||
def _required_text(payload: dict[str, Any], key: str, fallback_key: str | None = None) -> str:
|
||||
value = payload.get(key)
|
||||
if value is None and fallback_key is not None:
|
||||
|
||||
Reference in New Issue
Block a user