generated from coulomb/repo-seed
Add Inter-Hub migration bundle importer
This commit is contained in:
419
src/core_hub/migration.py
Normal file
419
src/core_hub/migration.py
Normal file
@@ -0,0 +1,419 @@
|
||||
import hashlib
|
||||
import json
|
||||
from collections.abc import Iterable
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from .models import (
|
||||
ApiConsumer,
|
||||
ApiKey,
|
||||
Hub,
|
||||
HubCapabilityManifest,
|
||||
InteractionEvent,
|
||||
MigrationRun,
|
||||
Widget,
|
||||
uuid_str,
|
||||
)
|
||||
|
||||
MIGRATION_SCHEMA_VERSION = "core-hub.migration.v1"
|
||||
COLLECTIONS = (
|
||||
"hubs",
|
||||
"hubCapabilityManifests",
|
||||
"apiConsumers",
|
||||
"apiKeys",
|
||||
"widgets",
|
||||
"interactionEvents",
|
||||
)
|
||||
SECRET_FIELD_NAMES = {"fullKey", "rawKey", "apiKey", "secret", "token"}
|
||||
|
||||
|
||||
def encode_body(payload: dict[str, Any]) -> str:
|
||||
return json.dumps(payload, sort_keys=True)
|
||||
|
||||
|
||||
def bundle_digest(bundle: dict[str, Any]) -> str:
|
||||
canonical = json.dumps(bundle, sort_keys=True, separators=(",", ":"))
|
||||
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def records_for(bundle: dict[str, Any], collection: str) -> list[dict[str, Any]]:
|
||||
records = bundle.get("records", {})
|
||||
if not isinstance(records, dict):
|
||||
return []
|
||||
values = records.get(collection, [])
|
||||
if not isinstance(values, list):
|
||||
return []
|
||||
return [record for record in values if isinstance(record, dict)]
|
||||
|
||||
|
||||
def _input_counts(bundle: dict[str, Any]) -> dict[str, dict[str, int]]:
|
||||
return {
|
||||
collection: {"input": len(records_for(bundle, collection))} for collection in COLLECTIONS
|
||||
}
|
||||
|
||||
|
||||
def _work_counts(bundle: dict[str, Any]) -> dict[str, dict[str, int]]:
|
||||
return {
|
||||
collection: {
|
||||
"input": len(records_for(bundle, collection)),
|
||||
"created": 0,
|
||||
"updated": 0,
|
||||
"skipped": 0,
|
||||
}
|
||||
for collection in COLLECTIONS
|
||||
}
|
||||
|
||||
|
||||
def _require(record: dict[str, Any], fields: Iterable[str], path: str, errors: list[str]) -> None:
|
||||
for field in fields:
|
||||
if record.get(field) in (None, ""):
|
||||
errors.append(f"{path}.{field} is required")
|
||||
|
||||
|
||||
def _note_duplicate(value: str | None, seen: set[str], path: str, errors: list[str]) -> None:
|
||||
if not value:
|
||||
return
|
||||
if value in seen:
|
||||
errors.append(f"duplicate {path}: {value}")
|
||||
seen.add(value)
|
||||
|
||||
|
||||
def _contains_secret_field(value: Any) -> bool:
|
||||
if isinstance(value, dict):
|
||||
for key, child in value.items():
|
||||
if key in SECRET_FIELD_NAMES:
|
||||
return True
|
||||
if _contains_secret_field(child):
|
||||
return True
|
||||
if isinstance(value, list):
|
||||
return any(_contains_secret_field(child) for child in value)
|
||||
return False
|
||||
|
||||
|
||||
def _relationship(status: str, name: str, message: str) -> dict[str, str]:
|
||||
return {"name": name, "status": status, "message": message}
|
||||
|
||||
|
||||
def validate_bundle(bundle: dict[str, Any]) -> dict[str, Any]:
|
||||
errors: list[str] = []
|
||||
warnings: list[str] = []
|
||||
relationships: list[dict[str, str]] = []
|
||||
|
||||
if bundle.get("schemaVersion") != MIGRATION_SCHEMA_VERSION:
|
||||
errors.append(
|
||||
f"schemaVersion must be {MIGRATION_SCHEMA_VERSION}, got {bundle.get('schemaVersion')!r}"
|
||||
)
|
||||
if not isinstance(bundle.get("records", {}), dict):
|
||||
errors.append("records must be an object")
|
||||
|
||||
hub_ids: set[str] = set()
|
||||
hub_slugs: set[str] = set()
|
||||
manifest_ids: set[str] = set()
|
||||
consumer_ids: set[str] = set()
|
||||
consumer_slugs: set[str] = set()
|
||||
widget_ids: set[str] = set()
|
||||
|
||||
for index, record in enumerate(records_for(bundle, "hubs")):
|
||||
path = f"records.hubs[{index}]"
|
||||
_require(record, ("slug", "name"), path, errors)
|
||||
_note_duplicate(record.get("id"), hub_ids, "hub id", errors)
|
||||
_note_duplicate(record.get("slug"), hub_slugs, "hub slug", errors)
|
||||
|
||||
for index, record in enumerate(records_for(bundle, "hubCapabilityManifests")):
|
||||
path = f"records.hubCapabilityManifests[{index}]"
|
||||
_require(record, ("id",), path, errors)
|
||||
_note_duplicate(record.get("id"), manifest_ids, "manifest id", errors)
|
||||
if not record.get("hubId") and not record.get("hubSlug"):
|
||||
errors.append(f"{path} must include hubId or hubSlug")
|
||||
elif record.get("hubId") not in hub_ids and record.get("hubSlug") not in hub_slugs:
|
||||
warnings.append(f"{path} references a hub outside this bundle")
|
||||
|
||||
for index, record in enumerate(records_for(bundle, "apiConsumers")):
|
||||
path = f"records.apiConsumers[{index}]"
|
||||
_require(record, ("name",), path, errors)
|
||||
if not record.get("id") and not record.get("slug"):
|
||||
errors.append(f"{path} must include id or slug for idempotent import")
|
||||
_note_duplicate(record.get("id"), consumer_ids, "api consumer id", errors)
|
||||
_note_duplicate(record.get("slug"), consumer_slugs, "api consumer slug", errors)
|
||||
manifest_id = record.get("hubCapabilityManifestId")
|
||||
if manifest_id and manifest_id not in manifest_ids:
|
||||
warnings.append(
|
||||
f"{path}.hubCapabilityManifestId references a manifest outside this bundle"
|
||||
)
|
||||
|
||||
for index, record in enumerate(records_for(bundle, "apiKeys")):
|
||||
path = f"records.apiKeys[{index}]"
|
||||
_require(record, ("id", "keyPrefix", "keyHash"), path, errors)
|
||||
if _contains_secret_field(record):
|
||||
errors.append(
|
||||
f"{path} contains a secret-like key field; import only hashes and prefixes"
|
||||
)
|
||||
if not record.get("apiConsumerId") and not record.get("apiConsumerSlug"):
|
||||
errors.append(f"{path} must include apiConsumerId or apiConsumerSlug")
|
||||
elif (
|
||||
record.get("apiConsumerId") not in consumer_ids
|
||||
and record.get("apiConsumerSlug") not in consumer_slugs
|
||||
):
|
||||
warnings.append(f"{path} references an API consumer outside this bundle")
|
||||
|
||||
for index, record in enumerate(records_for(bundle, "widgets")):
|
||||
path = f"records.widgets[{index}]"
|
||||
_require(record, ("id", "name"), path, errors)
|
||||
_note_duplicate(record.get("id"), widget_ids, "widget id", errors)
|
||||
if not record.get("hubId") and not record.get("hubSlug"):
|
||||
errors.append(f"{path} must include hubId or hubSlug")
|
||||
elif record.get("hubId") not in hub_ids and record.get("hubSlug") not in hub_slugs:
|
||||
warnings.append(f"{path} references a hub outside this bundle")
|
||||
|
||||
for index, record in enumerate(records_for(bundle, "interactionEvents")):
|
||||
path = f"records.interactionEvents[{index}]"
|
||||
_require(record, ("id", "eventType"), path, errors)
|
||||
if not record.get("widgetId"):
|
||||
errors.append(f"{path}.widgetId is required")
|
||||
elif record.get("widgetId") not in widget_ids:
|
||||
warnings.append(f"{path}.widgetId references a widget outside this bundle")
|
||||
|
||||
if errors:
|
||||
relationships.append(_relationship("fail", "bundle_integrity", "validation errors present"))
|
||||
else:
|
||||
relationships.append(
|
||||
_relationship("pass", "bundle_integrity", "required fields are present")
|
||||
)
|
||||
if warnings:
|
||||
relationships.append(
|
||||
_relationship(
|
||||
"warn", "external_references", "some references must resolve in the target DB"
|
||||
)
|
||||
)
|
||||
else:
|
||||
relationships.append(
|
||||
_relationship("pass", "external_references", "all references resolve within the bundle")
|
||||
)
|
||||
|
||||
return {
|
||||
"ok": not errors,
|
||||
"schemaVersion": bundle.get("schemaVersion"),
|
||||
"source": bundle.get("source", "unknown"),
|
||||
"bundleSha256": bundle_digest(bundle),
|
||||
"counts": _input_counts(bundle),
|
||||
"errors": errors,
|
||||
"warnings": warnings,
|
||||
"relationshipChecks": relationships,
|
||||
}
|
||||
|
||||
|
||||
async def _one_or_none(session: AsyncSession, statement: Any) -> Any | None:
|
||||
result = await session.execute(statement)
|
||||
return result.scalars().first()
|
||||
|
||||
|
||||
async def _hub_maps(session: AsyncSession) -> tuple[dict[str, Hub], dict[str, Hub]]:
|
||||
result = await session.execute(select(Hub))
|
||||
rows = list(result.scalars())
|
||||
return {row.id: row for row in rows}, {row.slug: row for row in rows}
|
||||
|
||||
|
||||
async def _manifest_maps(session: AsyncSession) -> dict[str, HubCapabilityManifest]:
|
||||
result = await session.execute(select(HubCapabilityManifest))
|
||||
rows = list(result.scalars())
|
||||
return {row.id: row for row in rows}
|
||||
|
||||
|
||||
async def _consumer_maps(
|
||||
session: AsyncSession,
|
||||
) -> tuple[dict[str, ApiConsumer], dict[str, ApiConsumer]]:
|
||||
result = await session.execute(select(ApiConsumer))
|
||||
rows = list(result.scalars())
|
||||
return {row.id: row for row in rows}, {row.slug: row for row in rows if row.slug}
|
||||
|
||||
|
||||
async def _widget_maps(session: AsyncSession) -> dict[str, Widget]:
|
||||
result = await session.execute(select(Widget))
|
||||
rows = list(result.scalars())
|
||||
return {row.id: row for row in rows}
|
||||
|
||||
|
||||
def _resolved_hub_id(
|
||||
record: dict[str, Any], hubs_by_id: dict[str, Hub], hubs_by_slug: dict[str, Hub]
|
||||
) -> str:
|
||||
if record.get("hubId"):
|
||||
return str(record["hubId"])
|
||||
return hubs_by_slug[str(record["hubSlug"])].id
|
||||
|
||||
|
||||
def _resolved_consumer_id(record: dict[str, Any], consumers_by_slug: dict[str, ApiConsumer]) -> str:
|
||||
if record.get("apiConsumerId"):
|
||||
return str(record["apiConsumerId"])
|
||||
return consumers_by_slug[str(record["apiConsumerSlug"])].id
|
||||
|
||||
|
||||
def _mark(counts: dict[str, dict[str, int]], collection: str, existing: Any | None) -> None:
|
||||
key = "updated" if existing else "created"
|
||||
counts[collection][key] += 1
|
||||
|
||||
|
||||
async def import_bundle(
|
||||
session: AsyncSession, bundle: dict[str, Any], *, dry_run: bool = False
|
||||
) -> dict[str, Any]:
|
||||
report = validate_bundle(bundle)
|
||||
report["dryRun"] = dry_run
|
||||
report["counts"] = _work_counts(bundle)
|
||||
if not report["ok"]:
|
||||
return report
|
||||
|
||||
hubs_by_id, hubs_by_slug = await _hub_maps(session)
|
||||
for record in records_for(bundle, "hubs"):
|
||||
existing = hubs_by_id.get(record.get("id")) or hubs_by_slug.get(record.get("slug"))
|
||||
_mark(report["counts"], "hubs", existing)
|
||||
if dry_run:
|
||||
continue
|
||||
row = existing or Hub(
|
||||
id=record.get("id") or uuid_str(),
|
||||
slug=record["slug"],
|
||||
name=record["name"],
|
||||
)
|
||||
row.slug = record["slug"]
|
||||
row.name = record["name"]
|
||||
row.domain = record.get("domain")
|
||||
row.hub_kind = record.get("hubKind")
|
||||
row.hub_family = record.get("hubFamily")
|
||||
row.vsm_function = record.get("vsmFunction")
|
||||
row.vsm_system = record.get("vsmSystem")
|
||||
row.status = record.get("status", "active")
|
||||
row.description = record.get("description")
|
||||
row.body_json = encode_body(record)
|
||||
session.add(row)
|
||||
if not dry_run:
|
||||
await session.flush()
|
||||
hubs_by_id, hubs_by_slug = await _hub_maps(session)
|
||||
|
||||
manifests_by_id = await _manifest_maps(session)
|
||||
for record in records_for(bundle, "hubCapabilityManifests"):
|
||||
existing = manifests_by_id.get(record.get("id"))
|
||||
_mark(report["counts"], "hubCapabilityManifests", existing)
|
||||
if dry_run:
|
||||
continue
|
||||
hub_id = _resolved_hub_id(record, hubs_by_id, hubs_by_slug)
|
||||
hub_slug = hubs_by_id[hub_id].slug if hub_id in hubs_by_id else record.get("hubSlug")
|
||||
row = existing or HubCapabilityManifest(id=record["id"], body_json="{}")
|
||||
row.hub_id = hub_id
|
||||
row.hub_slug = hub_slug
|
||||
row.manifest_version = record.get("manifestVersion", "0.1.0")
|
||||
row.status = record.get("status", "draft")
|
||||
row.body_json = encode_body(record)
|
||||
session.add(row)
|
||||
if not dry_run:
|
||||
await session.flush()
|
||||
|
||||
consumers_by_id, consumers_by_slug = await _consumer_maps(session)
|
||||
for record in records_for(bundle, "apiConsumers"):
|
||||
existing = consumers_by_id.get(record.get("id")) or consumers_by_slug.get(
|
||||
record.get("slug")
|
||||
)
|
||||
_mark(report["counts"], "apiConsumers", existing)
|
||||
if dry_run:
|
||||
continue
|
||||
row = existing or ApiConsumer(
|
||||
id=record.get("id") or uuid_str(),
|
||||
slug=record.get("slug"),
|
||||
name=record["name"],
|
||||
)
|
||||
row.slug = record.get("slug")
|
||||
row.name = record["name"]
|
||||
row.description = record.get("description")
|
||||
row.hub_capability_manifest_id = record.get("hubCapabilityManifestId")
|
||||
row.rate_limit_per_minute = record.get("rateLimitPerMinute")
|
||||
row.quota_per_day = record.get("quotaPerDay")
|
||||
row.key_prefix = record.get("keyPrefix")
|
||||
row.key_hash = record.get("keyHash")
|
||||
row.status = record.get("status", "active")
|
||||
row.body_json = encode_body(record)
|
||||
session.add(row)
|
||||
if not dry_run:
|
||||
await session.flush()
|
||||
consumers_by_id, consumers_by_slug = await _consumer_maps(session)
|
||||
|
||||
for record in records_for(bundle, "apiKeys"):
|
||||
existing = await session.get(ApiKey, record.get("id"))
|
||||
if existing is None:
|
||||
existing = await _one_or_none(
|
||||
session, select(ApiKey).where(ApiKey.key_hash == record["keyHash"])
|
||||
)
|
||||
_mark(report["counts"], "apiKeys", existing)
|
||||
if dry_run:
|
||||
continue
|
||||
row = existing or ApiKey(
|
||||
id=record["id"],
|
||||
api_consumer_id=_resolved_consumer_id(record, consumers_by_slug),
|
||||
)
|
||||
row.api_consumer_id = _resolved_consumer_id(record, consumers_by_slug)
|
||||
row.key_prefix = record["keyPrefix"]
|
||||
row.key_hash = record["keyHash"]
|
||||
row.scopes = record.get("scopes")
|
||||
row.status = record.get("status", "active")
|
||||
session.add(row)
|
||||
|
||||
widgets_by_id = await _widget_maps(session)
|
||||
for record in records_for(bundle, "widgets"):
|
||||
existing = widgets_by_id.get(record.get("id"))
|
||||
_mark(report["counts"], "widgets", existing)
|
||||
if dry_run:
|
||||
continue
|
||||
row = existing or Widget(
|
||||
id=record["id"],
|
||||
hub_id=_resolved_hub_id(record, hubs_by_id, hubs_by_slug),
|
||||
name=record["name"],
|
||||
)
|
||||
row.hub_id = _resolved_hub_id(record, hubs_by_id, hubs_by_slug)
|
||||
row.name = record["name"]
|
||||
row.widget_type = record.get("widgetType")
|
||||
row.capability_ref = record.get("capabilityRef")
|
||||
row.view_context = record.get("viewContext")
|
||||
row.policy_scope = record.get("policyScope")
|
||||
row.status = record.get("status", "active")
|
||||
row.body_json = encode_body(record)
|
||||
session.add(row)
|
||||
if not dry_run:
|
||||
await session.flush()
|
||||
|
||||
for record in records_for(bundle, "interactionEvents"):
|
||||
existing = await session.get(InteractionEvent, record.get("id"))
|
||||
_mark(report["counts"], "interactionEvents", existing)
|
||||
if dry_run:
|
||||
continue
|
||||
row = existing or InteractionEvent(
|
||||
id=record["id"],
|
||||
widget_id=record["widgetId"],
|
||||
event_type=record["eventType"],
|
||||
)
|
||||
row.widget_id = record["widgetId"]
|
||||
row.event_type = record["eventType"]
|
||||
row.view_context = record.get("viewContext")
|
||||
row.metadata_json = encode_body(record.get("metadata", {}))
|
||||
row.body_json = encode_body(record)
|
||||
session.add(row)
|
||||
|
||||
if not dry_run:
|
||||
run = MigrationRun(
|
||||
source=report["source"],
|
||||
schema_version=report["schemaVersion"],
|
||||
bundle_sha256=report["bundleSha256"],
|
||||
dry_run=False,
|
||||
status="imported",
|
||||
counts_json=encode_body(report["counts"]),
|
||||
diagnostics_json=encode_body(
|
||||
{
|
||||
"errors": report["errors"],
|
||||
"warnings": report["warnings"],
|
||||
"relationshipChecks": report["relationshipChecks"],
|
||||
}
|
||||
),
|
||||
)
|
||||
session.add(run)
|
||||
await session.commit()
|
||||
await session.refresh(run)
|
||||
report["migrationRunId"] = run.id
|
||||
|
||||
return report
|
||||
@@ -1,7 +1,7 @@
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func
|
||||
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, func
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from .db import Base
|
||||
@@ -102,3 +102,17 @@ class InteractionEvent(Base):
|
||||
metadata_json: Mapped[str] = mapped_column(Text(), default="{}")
|
||||
body_json: Mapped[str] = mapped_column(Text(), default="{}")
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
|
||||
|
||||
|
||||
class MigrationRun(Base):
|
||||
__tablename__ = "migration_runs"
|
||||
|
||||
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_str)
|
||||
source: Mapped[str] = mapped_column(String(120), index=True)
|
||||
schema_version: Mapped[str] = mapped_column(String(80))
|
||||
bundle_sha256: Mapped[str] = mapped_column(String(64), index=True)
|
||||
dry_run: Mapped[bool] = mapped_column(Boolean(), default=False)
|
||||
status: Mapped[str] = mapped_column(String(40), default="imported")
|
||||
counts_json: Mapped[str] = mapped_column(Text(), default="{}")
|
||||
diagnostics_json: Mapped[str] = mapped_column(Text(), default="{}")
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
|
||||
|
||||
Reference in New Issue
Block a user