tegwick
9751927d38
All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 38s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b22-9638-76d2-bbff-b7ea1770b118
606 lines
23 KiB
Python
606 lines
23 KiB
Python
"""Transactional persistence for SQLite development and PostgreSQL production."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
import uuid
|
|
from collections import defaultdict
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func, insert, select, update
|
|
from sqlalchemy.engine import RowMapping
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from sbom_nexus.database import (
|
|
create_database_engine,
|
|
metadata,
|
|
operation_receipts,
|
|
repositories,
|
|
snapshots,
|
|
)
|
|
from sbom_nexus.database import entries as entry_table
|
|
from sbom_nexus.scanner import is_copyleft
|
|
|
|
SUCCESS_STATUSES = frozenset({"ingested", "imported"})
|
|
|
|
|
|
class OperationKeyConflict(ValueError):
|
|
"""An operation key was reused for a different mutation request."""
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def parse_datetime(value: datetime | str) -> datetime:
|
|
parsed = (
|
|
datetime.fromisoformat(value.replace("Z", "+00:00")) if isinstance(value, str) else value
|
|
)
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=UTC)
|
|
return parsed.astimezone(UTC)
|
|
|
|
|
|
def datetime_text(value: datetime | str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return parse_datetime(value).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
class Store:
|
|
def __init__(
|
|
self,
|
|
database_target: str | Path,
|
|
*,
|
|
database_url_file: str | Path | None = None,
|
|
) -> None:
|
|
self.engine = create_database_engine(
|
|
database_target, database_url_file=database_url_file
|
|
)
|
|
|
|
@property
|
|
def dialect(self) -> str:
|
|
return self.engine.dialect.name
|
|
|
|
def init_schema(self) -> None:
|
|
metadata.create_all(self.engine)
|
|
|
|
def health(self) -> None:
|
|
with self.engine.connect() as connection:
|
|
connection.execute(select(func.count()).select_from(repositories))
|
|
|
|
def upsert_repository(
|
|
self,
|
|
slug: str,
|
|
*,
|
|
checkout_path: str | None = None,
|
|
source_ref: dict[str, Any] | None = None,
|
|
active: bool = True,
|
|
last_attempt_at: datetime | str | None = None,
|
|
last_success_at: datetime | str | None = None,
|
|
) -> dict[str, Any]:
|
|
now = utc_now()
|
|
attempt = parse_datetime(last_attempt_at) if last_attempt_at else None
|
|
success = parse_datetime(last_success_at) if last_success_at else None
|
|
with self.engine.begin() as connection:
|
|
existing = (
|
|
connection.execute(select(repositories).where(repositories.c.slug == slug))
|
|
.mappings()
|
|
.one_or_none()
|
|
)
|
|
if existing:
|
|
connection.execute(
|
|
update(repositories)
|
|
.where(repositories.c.slug == slug)
|
|
.values(
|
|
checkout_path=checkout_path,
|
|
source_ref_json=source_ref,
|
|
active=active,
|
|
last_attempt_at=attempt or existing["last_attempt_at"],
|
|
last_success_at=success or existing["last_success_at"],
|
|
updated_at=now,
|
|
)
|
|
)
|
|
else:
|
|
connection.execute(
|
|
insert(repositories).values(
|
|
id=str(uuid.uuid4()),
|
|
slug=slug,
|
|
checkout_path=checkout_path,
|
|
source_ref_json=source_ref,
|
|
active=active,
|
|
last_attempt_at=attempt,
|
|
last_success_at=success,
|
|
last_status=None,
|
|
last_source=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
row = (
|
|
connection.execute(select(repositories).where(repositories.c.slug == slug))
|
|
.mappings()
|
|
.one()
|
|
)
|
|
return self._repository_dict(row)
|
|
|
|
def get_repository(self, slug: str) -> dict[str, Any] | None:
|
|
with self.engine.connect() as connection:
|
|
row = (
|
|
connection.execute(select(repositories).where(repositories.c.slug == slug))
|
|
.mappings()
|
|
.one_or_none()
|
|
)
|
|
return self._repository_dict(row) if row else None
|
|
|
|
def list_repositories(self) -> list[dict[str, Any]]:
|
|
with self.engine.connect() as connection:
|
|
rows = (
|
|
connection.execute(select(repositories).order_by(repositories.c.slug))
|
|
.mappings()
|
|
.all()
|
|
)
|
|
return [self._repository_dict(row) for row in rows]
|
|
|
|
def replay_operation(
|
|
self, operation_key: str | None, request_fingerprint: str | None
|
|
) -> dict[str, Any] | None:
|
|
if operation_key is None:
|
|
return None
|
|
if request_fingerprint is None:
|
|
raise ValueError("request_fingerprint is required with operation_key")
|
|
with self.engine.connect() as connection:
|
|
receipt = (
|
|
connection.execute(
|
|
select(operation_receipts).where(
|
|
operation_receipts.c.operation_key == operation_key
|
|
)
|
|
)
|
|
.mappings()
|
|
.one_or_none()
|
|
)
|
|
if receipt is None:
|
|
return None
|
|
if receipt["request_fingerprint"] != request_fingerprint:
|
|
raise OperationKeyConflict("operation key was already used for a different request")
|
|
snapshot = (
|
|
connection.execute(
|
|
select(snapshots).where(snapshots.c.id == receipt["snapshot_id"])
|
|
)
|
|
.mappings()
|
|
.one()
|
|
)
|
|
return self._snapshot_dict(snapshot)
|
|
|
|
def record_snapshot(
|
|
self,
|
|
repo_slug: str,
|
|
*,
|
|
entries: list[dict[str, Any]],
|
|
status: str,
|
|
source: str,
|
|
snapshot_at: datetime | str | None = None,
|
|
source_revision: str | None = None,
|
|
source_provenance: dict[str, Any] | None = None,
|
|
sources: list[dict[str, Any]] | None = None,
|
|
errors: list[dict[str, Any]] | None = None,
|
|
snapshot_id: str | None = None,
|
|
legacy_id: str | None = None,
|
|
operation_key: str | None = None,
|
|
request_fingerprint: str | None = None,
|
|
) -> tuple[dict[str, Any], bool]:
|
|
if bool(operation_key) != bool(request_fingerprint):
|
|
raise ValueError("operation_key and request_fingerprint must be supplied together")
|
|
timestamp = parse_datetime(snapshot_at or utc_now())
|
|
created_at = utc_now()
|
|
try:
|
|
with self.engine.begin() as connection:
|
|
return self._record_snapshot_transaction(
|
|
connection,
|
|
repo_slug=repo_slug,
|
|
entries=entries,
|
|
status=status,
|
|
source=source,
|
|
timestamp=timestamp,
|
|
created_at=created_at,
|
|
source_revision=source_revision,
|
|
source_provenance=source_provenance,
|
|
sources=sources,
|
|
errors=errors,
|
|
snapshot_id=snapshot_id,
|
|
legacy_id=legacy_id,
|
|
operation_key=operation_key,
|
|
request_fingerprint=request_fingerprint,
|
|
)
|
|
except IntegrityError:
|
|
if not operation_key or not request_fingerprint:
|
|
raise
|
|
for _attempt in range(20):
|
|
replay = self.replay_operation(operation_key, request_fingerprint)
|
|
if replay is not None:
|
|
return replay, False
|
|
time.sleep(0.05)
|
|
raise
|
|
|
|
def _record_snapshot_transaction(
|
|
self,
|
|
connection,
|
|
*,
|
|
repo_slug: str,
|
|
entries: list[dict[str, Any]],
|
|
status: str,
|
|
source: str,
|
|
timestamp: datetime,
|
|
created_at: datetime,
|
|
source_revision: str | None,
|
|
source_provenance: dict[str, Any] | None,
|
|
sources: list[dict[str, Any]] | None,
|
|
errors: list[dict[str, Any]] | None,
|
|
snapshot_id: str | None,
|
|
legacy_id: str | None,
|
|
operation_key: str | None,
|
|
request_fingerprint: str | None,
|
|
) -> tuple[dict[str, Any], bool]:
|
|
repo = (
|
|
connection.execute(select(repositories).where(repositories.c.slug == repo_slug))
|
|
.mappings()
|
|
.one_or_none()
|
|
)
|
|
if repo is None:
|
|
raise KeyError(repo_slug)
|
|
if operation_key:
|
|
receipt = (
|
|
connection.execute(
|
|
select(operation_receipts).where(
|
|
operation_receipts.c.operation_key == operation_key
|
|
)
|
|
)
|
|
.mappings()
|
|
.one_or_none()
|
|
)
|
|
if receipt:
|
|
if receipt["request_fingerprint"] != request_fingerprint:
|
|
raise OperationKeyConflict(
|
|
"operation key was already used for a different request"
|
|
)
|
|
existing = (
|
|
connection.execute(
|
|
select(snapshots).where(snapshots.c.id == receipt["snapshot_id"])
|
|
)
|
|
.mappings()
|
|
.one()
|
|
)
|
|
return self._snapshot_dict(existing), False
|
|
if legacy_id:
|
|
existing = (
|
|
connection.execute(select(snapshots).where(snapshots.c.legacy_id == legacy_id))
|
|
.mappings()
|
|
.one_or_none()
|
|
)
|
|
if existing:
|
|
return self._snapshot_dict(existing), False
|
|
|
|
new_snapshot_id = snapshot_id or str(uuid.uuid4())
|
|
connection.execute(
|
|
insert(snapshots).values(
|
|
id=new_snapshot_id,
|
|
repo_id=repo["id"],
|
|
snapshot_at=timestamp,
|
|
source=source,
|
|
status=status,
|
|
entry_count=len(entries),
|
|
source_revision=source_revision,
|
|
source_provenance_json=source_provenance,
|
|
sources_json=sources or [],
|
|
errors_json=errors or [],
|
|
legacy_id=legacy_id,
|
|
created_at=created_at,
|
|
)
|
|
)
|
|
if entries:
|
|
connection.execute(
|
|
insert(entry_table),
|
|
[
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"repo_id": repo["id"],
|
|
"snapshot_id": new_snapshot_id,
|
|
"package_name": entry["package_name"],
|
|
"package_version": entry.get("package_version"),
|
|
"ecosystem": entry["ecosystem"],
|
|
"license_spdx": entry.get("license_spdx"),
|
|
"is_direct": bool(entry.get("is_direct", True)),
|
|
"is_dev": bool(entry.get("is_dev", False)),
|
|
"source_path": entry.get("source_path"),
|
|
"created_at": created_at,
|
|
}
|
|
for entry in entries
|
|
],
|
|
)
|
|
|
|
previous_attempt = repo["last_attempt_at"]
|
|
is_latest = previous_attempt is None or timestamp >= parse_datetime(previous_attempt)
|
|
if is_latest:
|
|
last_success = repo["last_success_at"]
|
|
if status in SUCCESS_STATUSES:
|
|
last_success = timestamp
|
|
connection.execute(
|
|
update(repositories)
|
|
.where(repositories.c.id == repo["id"])
|
|
.values(
|
|
last_attempt_at=timestamp,
|
|
last_success_at=last_success,
|
|
last_status=status,
|
|
last_source=source,
|
|
updated_at=created_at,
|
|
)
|
|
)
|
|
if operation_key:
|
|
connection.execute(
|
|
insert(operation_receipts).values(
|
|
operation_key=operation_key,
|
|
request_fingerprint=request_fingerprint,
|
|
snapshot_id=new_snapshot_id,
|
|
created_at=created_at,
|
|
)
|
|
)
|
|
row = (
|
|
connection.execute(select(snapshots).where(snapshots.c.id == new_snapshot_id))
|
|
.mappings()
|
|
.one()
|
|
)
|
|
return self._snapshot_dict(row), True
|
|
|
|
def list_snapshots(self, repo_slug: str | None = None) -> list[dict[str, Any]]:
|
|
statement = (
|
|
select(snapshots, repositories.c.slug.label("repo_slug"))
|
|
.join(repositories, repositories.c.id == snapshots.c.repo_id)
|
|
.order_by(snapshots.c.snapshot_at.desc(), snapshots.c.created_at.desc())
|
|
)
|
|
if repo_slug:
|
|
statement = statement.where(repositories.c.slug == repo_slug)
|
|
with self.engine.connect() as connection:
|
|
rows = connection.execute(statement).mappings().all()
|
|
return [self._snapshot_dict(row) for row in rows]
|
|
|
|
def get_snapshot(self, snapshot_id: str) -> dict[str, Any] | None:
|
|
statement = (
|
|
select(snapshots, repositories.c.slug.label("repo_slug"))
|
|
.join(repositories, repositories.c.id == snapshots.c.repo_id)
|
|
.where(snapshots.c.id == snapshot_id)
|
|
)
|
|
with self.engine.connect() as connection:
|
|
row = connection.execute(statement).mappings().one_or_none()
|
|
if row is None:
|
|
return None
|
|
entry_rows = (
|
|
connection.execute(
|
|
select(entry_table)
|
|
.where(entry_table.c.snapshot_id == snapshot_id)
|
|
.order_by(entry_table.c.package_name)
|
|
)
|
|
.mappings()
|
|
.all()
|
|
)
|
|
result = self._snapshot_dict(row)
|
|
result["entries"] = []
|
|
for entry in entry_rows:
|
|
rendered = self._entry_dict(entry)
|
|
rendered["snapshot_at"] = result["snapshot_at"]
|
|
result["entries"].append(rendered)
|
|
return result
|
|
|
|
def latest_entries(
|
|
self,
|
|
*,
|
|
repo_slug: str | None = None,
|
|
ecosystem: str | None = None,
|
|
license_spdx: str | None = None,
|
|
is_direct: bool | None = None,
|
|
is_dev: bool | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
ranked = self._ranked_snapshots()
|
|
statement = (
|
|
select(
|
|
entry_table,
|
|
repositories.c.slug.label("repo_slug"),
|
|
snapshots.c.snapshot_at.label("entry_snapshot_at"),
|
|
)
|
|
.join(repositories, repositories.c.id == entry_table.c.repo_id)
|
|
.join(snapshots, snapshots.c.id == entry_table.c.snapshot_id)
|
|
.join(ranked, ranked.c.id == entry_table.c.snapshot_id)
|
|
.where(ranked.c.snapshot_rank == 1)
|
|
)
|
|
if repo_slug:
|
|
statement = statement.where(repositories.c.slug == repo_slug)
|
|
for column, value in (
|
|
(entry_table.c.ecosystem, ecosystem),
|
|
(entry_table.c.license_spdx, license_spdx),
|
|
(entry_table.c.is_direct, is_direct),
|
|
(entry_table.c.is_dev, is_dev),
|
|
):
|
|
if value is not None:
|
|
statement = statement.where(column == value)
|
|
statement = statement.order_by(entry_table.c.package_name, repositories.c.slug)
|
|
with self.engine.connect() as connection:
|
|
rows = connection.execute(statement).mappings().all()
|
|
return [self._entry_dict(row) for row in rows]
|
|
|
|
def repository_view(self, repo_slug: str) -> dict[str, Any] | None:
|
|
repo = self.get_repository(repo_slug)
|
|
if repo is None:
|
|
return None
|
|
current_entries = self.latest_entries(repo_slug=repo_slug)
|
|
history = self.list_snapshots(repo_slug)
|
|
return {
|
|
"repo_slug": repo_slug,
|
|
"last_sbom_at": repo["last_attempt_at"],
|
|
"last_attempt_at": repo["last_attempt_at"],
|
|
"last_success_at": repo["last_success_at"],
|
|
"last_status": repo["last_status"],
|
|
"entry_count": len(current_entries),
|
|
"snapshot_id": history[0]["id"] if history else None,
|
|
"entries": current_entries,
|
|
}
|
|
|
|
def licence_report(self) -> dict[str, Any]:
|
|
current_entries = self.latest_entries()
|
|
groups: dict[str | None, dict[str, Any]] = defaultdict(lambda: {"count": 0, "repos": set()})
|
|
risks: list[dict[str, Any]] = []
|
|
for entry in current_entries:
|
|
license_id = entry.get("license_spdx")
|
|
groups[license_id]["count"] += 1
|
|
groups[license_id]["repos"].add(entry["repo_slug"])
|
|
if is_copyleft(license_id) and entry["is_direct"] and not entry["is_dev"]:
|
|
risks.append(
|
|
{
|
|
"repo_slug": entry["repo_slug"],
|
|
"package_name": entry["package_name"],
|
|
"package_version": entry.get("package_version"),
|
|
"license_spdx": license_id,
|
|
"source_path": entry.get("source_path"),
|
|
}
|
|
)
|
|
sorted_groups = sorted(groups.items(), key=lambda item: (-item[1]["count"], item[0] or ""))
|
|
return {
|
|
"groups": [
|
|
{
|
|
"license_spdx": license_id,
|
|
"count": data["count"],
|
|
"repos": sorted(data["repos"]),
|
|
"is_copyleft": is_copyleft(license_id),
|
|
}
|
|
for license_id, data in sorted_groups
|
|
],
|
|
"copyleft_direct_prod": risks,
|
|
"copyleft_direct_count": len(risks),
|
|
"signal_qualification": "substring triage; not legal advice or full SPDX evaluation",
|
|
}
|
|
|
|
def catch_up(
|
|
self,
|
|
*,
|
|
limit: int = 3,
|
|
stale_days: int = 30,
|
|
now: datetime | None = None,
|
|
) -> dict[str, Any]:
|
|
effective_limit = max(1, min(25, int(limit)))
|
|
current = parse_datetime(now or utc_now())
|
|
cutoff = current - timedelta(days=stale_days)
|
|
active_repositories = [repo for repo in self.list_repositories() if repo["active"]]
|
|
never = [repo for repo in active_repositories if repo["last_attempt_at"] is None]
|
|
stale = [
|
|
repo
|
|
for repo in active_repositories
|
|
if repo["last_attempt_at"] is None or parse_datetime(repo["last_attempt_at"]) < cutoff
|
|
]
|
|
stale.sort(
|
|
key=lambda repo: (
|
|
repo["last_attempt_at"] is not None,
|
|
repo["last_attempt_at"] or "",
|
|
repo["slug"],
|
|
)
|
|
)
|
|
selected = stale[:effective_limit]
|
|
return {
|
|
"repos": [self._catch_up_repository(repo, current) for repo in selected],
|
|
"selected_count": len(selected),
|
|
"stale_count": len(stale),
|
|
"never_count": len(never),
|
|
"total_count": len(active_repositories),
|
|
"limit": effective_limit,
|
|
"stale_after_days": stale_days,
|
|
"evaluated_at": datetime_text(current),
|
|
}
|
|
|
|
@staticmethod
|
|
def _ranked_snapshots():
|
|
return select(
|
|
snapshots.c.id,
|
|
func.row_number()
|
|
.over(
|
|
partition_by=snapshots.c.repo_id,
|
|
order_by=(snapshots.c.snapshot_at.desc(), snapshots.c.created_at.desc()),
|
|
)
|
|
.label("snapshot_rank"),
|
|
).subquery("ranked_snapshots")
|
|
|
|
@staticmethod
|
|
def _repository_dict(row: RowMapping) -> dict[str, Any]:
|
|
return {
|
|
"id": row["id"],
|
|
"slug": row["slug"],
|
|
"checkout_path": row["checkout_path"],
|
|
"source_ref": row["source_ref_json"],
|
|
"active": bool(row["active"]),
|
|
"last_attempt_at": datetime_text(row["last_attempt_at"]),
|
|
"last_success_at": datetime_text(row["last_success_at"]),
|
|
"last_status": row["last_status"],
|
|
"last_source": row["last_source"],
|
|
"created_at": datetime_text(row["created_at"]),
|
|
"updated_at": datetime_text(row["updated_at"]),
|
|
}
|
|
|
|
@staticmethod
|
|
def _snapshot_dict(row: RowMapping) -> dict[str, Any]:
|
|
result = {
|
|
"id": row["id"],
|
|
"repo_id": row["repo_id"],
|
|
"snapshot_at": datetime_text(row["snapshot_at"]),
|
|
"source": row["source"],
|
|
"status": row["status"],
|
|
"entry_count": row["entry_count"],
|
|
"source_revision": row["source_revision"],
|
|
"source_provenance": row["source_provenance_json"],
|
|
"sources": row["sources_json"] or [],
|
|
"errors": row["errors_json"] or [],
|
|
"legacy_id": row["legacy_id"],
|
|
"created_at": datetime_text(row["created_at"]),
|
|
}
|
|
if "repo_slug" in row:
|
|
result["repo_slug"] = row["repo_slug"]
|
|
return result
|
|
|
|
@staticmethod
|
|
def _entry_dict(row: RowMapping) -> dict[str, Any]:
|
|
result = {
|
|
"id": row["id"],
|
|
"repo_id": row["repo_id"],
|
|
"snapshot_id": row["snapshot_id"],
|
|
"package_name": row["package_name"],
|
|
"package_version": row["package_version"],
|
|
"ecosystem": row["ecosystem"],
|
|
"license_spdx": row["license_spdx"],
|
|
"is_direct": bool(row["is_direct"]),
|
|
"is_dev": bool(row["is_dev"]),
|
|
"source_path": row["source_path"],
|
|
"created_at": datetime_text(row["created_at"]),
|
|
}
|
|
if "repo_slug" in row:
|
|
result["repo_slug"] = row["repo_slug"]
|
|
if "entry_snapshot_at" in row:
|
|
result["snapshot_at"] = datetime_text(row["entry_snapshot_at"])
|
|
return result
|
|
|
|
@staticmethod
|
|
def _catch_up_repository(repo: dict[str, Any], now: datetime) -> dict[str, Any]:
|
|
attempt = repo["last_attempt_at"]
|
|
age_days = 9999 if attempt is None else max(0, (now - parse_datetime(attempt)).days)
|
|
checkout_path = repo["checkout_path"]
|
|
source_ref = repo.get("source_ref")
|
|
return {
|
|
"repo_slug": repo["slug"],
|
|
"last_sbom_at": attempt,
|
|
"last_attempt_at": attempt,
|
|
"last_success_at": repo["last_success_at"],
|
|
"sbom_age_days": age_days,
|
|
"has_sbom": repo["last_success_at"] is not None,
|
|
"checkout_available": bool(
|
|
source_ref or (checkout_path and Path(checkout_path).is_dir())
|
|
),
|
|
"source_available": source_ref is not None,
|
|
"source_ref": source_ref,
|
|
"last_status": repo["last_status"],
|
|
}
|