feat: topology boostrap and persistance (9ff67cb7) · Commits · OOP / code / service-resource-manager

src/srm/adapters/database/repos/topology.py

+46 −22

Changes for src/srm/adapters/database/repos/topology.py: 46 added lines, 22 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from typing import Any
		from uuid import UUID

		import structlog
		@@ -41,20 +42,33 @@ def _log_integrity_error(exc: IntegrityError, *, entity: str, operation: str) ->
		)


		def _zone_options() -> Any:
		return (
		selectinload(ZoneRow.domains)
		.selectinload(DomainRow.capabilities)
		.selectinload(CapabilityRow.control_path_bindings)
		)


		def _domain_options() -> Any:
		return selectinload(DomainRow.capabilities).selectinload(CapabilityRow.control_path_bindings)


		def _capability_options() -> Any:
		return selectinload(CapabilityRow.control_path_bindings)


		class SqlZoneRepository(ZoneRepository):
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def get_by_id(self, id: UUID) -> Zone \| None:
		stmt = (
		select(ZoneRow)
		.options(
		selectinload(ZoneRow.domains)
		.selectinload(DomainRow.capabilities)
		.selectinload(CapabilityRow.control_path_bindings)
		)
		.where(ZoneRow.id == id)
		)
		stmt = select(ZoneRow).options(_zone_options()).where(ZoneRow.id == id)
		row = await self._session.scalar(stmt)
		return ZoneMapper.to_domain(row) if row is not None else None

		async def get_by_ref(self, ref: str) -> Zone \| None:
		stmt = select(ZoneRow).options(_zone_options()).where(ZoneRow.ref == ref)
		row = await self._session.scalar(stmt)
		return ZoneMapper.to_domain(row) if row is not None else None

		@@ -69,11 +83,7 @@ class SqlZoneRepository(ZoneRepository):
		) -> list[Zone]:
		stmt = (
		select(ZoneRow)
		.options(
		selectinload(ZoneRow.domains)
		.selectinload(DomainRow.capabilities)
		.selectinload(CapabilityRow.control_path_bindings)
		)
		.options(_zone_options())
		.where(ZoneRow.kind == ZoneKind.RESOURCE)
		.order_by(ZoneRow.ref, ZoneRow.id)
		)
		@@ -106,14 +116,15 @@ class SqlDomainRepository(DomainRepository):
		self._session = session

		async def get_by_id(self, id: UUID) -> Domain \| None:
		stmt = select(DomainRow).options(_domain_options()).where(DomainRow.id == id)
		row = await self._session.scalar(stmt)
		return DomainMapper.to_domain(row) if row is not None else None

		async def get_by_ref(self, zone_id: UUID, ref: str) -> Domain \| None:
		stmt = (
		select(DomainRow)
		.options(
		selectinload(DomainRow.capabilities).selectinload(
		CapabilityRow.control_path_bindings
		)
		)
		.where(DomainRow.id == id)
		.options(_domain_options())
		.where(DomainRow.zone_id == zone_id, DomainRow.ref == ref)
		)
		row = await self._session.scalar(stmt)
		return DomainMapper.to_domain(row) if row is not None else None
		@@ -138,10 +149,15 @@ class SqlCapabilityRepository(CapabilityRepository):
		self._session = session

		async def get_by_id(self, id: UUID) -> Capability \| None:
		stmt = select(CapabilityRow).options(_capability_options()).where(CapabilityRow.id == id)
		row = await self._session.scalar(stmt)
		return CapabilityMapper.to_domain(row) if row is not None else None

		async def get_by_ref(self, domain_id: UUID, ref: str) -> Capability \| None:
		stmt = (
		select(CapabilityRow)
		.options(selectinload(CapabilityRow.control_path_bindings))
		.where(CapabilityRow.id == id)
		.options(_capability_options())
		.where(CapabilityRow.domain_id == domain_id, CapabilityRow.ref == ref)
		)
		row = await self._session.scalar(stmt)
		return CapabilityMapper.to_domain(row) if row is not None else None
		@@ -170,6 +186,14 @@ class SqlControlPathBindingRepository(ControlPathBindingRepository):
		row = await self._session.scalar(stmt)
		return ControlPathBindingMapper.to_domain(row) if row is not None else None

		async def get_by_ref(self, capability_id: UUID, ref: str) -> ControlPathBinding \| None:
		stmt = select(ControlPathBindingRow).where(
		ControlPathBindingRow.capability_id == capability_id,
		ControlPathBindingRow.ref == ref,
		)
		row = await self._session.scalar(stmt)
		return ControlPathBindingMapper.to_domain(row) if row is not None else None

		async def create(self, control_path_binding: ControlPathBinding) -> ControlPathBinding:
		row = ControlPathBindingMapper.to_row(control_path_binding)
		self._session.add(row)

src/srm/adapters/transformation_functions/tfsdk_topology.py

+9 −0

Changes for src/srm/adapters/transformation_functions/tfsdk_topology.py: 9 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -8,6 +8,7 @@ from __future__ import annotations

		from typing import Any

		import structlog
		from openop_tfsdk.contract.interfaces import TopologyDiscoveryAdapter
		from openop_tfsdk.contract.topology import BootstrapTopology

		@@ -29,6 +30,8 @@ from srm.domain.ports.transformation_functions import (
		TransformationFunctionZone,
		)

		logger = structlog.get_logger(__name__)

		# TF-SDK zone state uses "offline"; SRM ZoneState uses "inactive".
		_ZONE_STATE_MAP: dict[str, ZoneState] = {
		"active": ZoneState.ACTIVE,
		@@ -171,6 +174,12 @@ class TfSdkTopologyDiscoveryBridge(TopologyDiscoveryPort):
		for capability in self._adapter.get_capabilities(domain_ref=domain_ref, kind=kind_filter):
		converted = _map_capability(capability)
		if converted is None:
		payload = capability.model_dump(mode="json")
		logger.warning(
		"bootstrap_capability_kind_unknown",
		ref=payload.get("ref"),
		kind=payload.get("kind"),
		)
		continue
		if kind is not None and converted.kind != kind:
		continue

src/srm/application/bootstrap/topology.py

0 → 100644

+228 −0

Changes for src/srm/application/bootstrap/topology.py: 228 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Materialize TF-SDK bootstrap topology into SRM's topology tables."""

		from __future__ import annotations

		from typing import Any
		from uuid import NAMESPACE_URL, UUID, uuid5

		import structlog

		from srm.domain.models.topology import (
		Capability,
		ControlPathBinding,
		Domain,
		Zone,
		)
		from srm.domain.ports.database.topology import (
		CapabilityRepository,
		ControlPathBindingRepository,
		DomainRepository,
		ZoneRepository,
		)
		from srm.domain.ports.transformation_functions import (
		TopologyDiscoveryPort,
		TransformationFunctionCapability,
		TransformationFunctionControlPathBinding,
		TransformationFunctionDomain,
		TransformationFunctionZone,
		)

		logger: structlog.BoundLogger = structlog.get_logger(__name__)


		class BootstrapTopologyError(Exception):
		"""Bootstrap YAML is invalid for materialization into SRM."""


		def well_known_zone_id(ref: str) -> UUID:
		"""Deterministic zone.id (CAMARA edgeCloudZoneId) from bootstrap ref."""
		return uuid5(NAMESPACE_URL, f"https://etsi.org/sdg/oop/zone/{ref}")


		def _zone_from(tf_zone: TransformationFunctionZone) -> Zone:
		metadata: dict[str, Any] = dict(tf_zone.metadata or {})

		if tf_zone.location:
		existing_location = metadata.get("location")
		existing: dict[str, Any] = (
		dict(existing_location) if isinstance(existing_location, dict) else {}
		)
		metadata["location"] = {existing, dict(tf_zone.location)}

		location = metadata.get("location")
		if isinstance(location, dict) and "region" not in location:
		labels = metadata.get("labels")
		if isinstance(labels, dict) and labels.get("region") is not None:
		metadata["location"] = {**location, "region": labels["region"]}
		elif not isinstance(location, dict):
		labels = metadata.get("labels")
		if isinstance(labels, dict) and labels.get("region") is not None:
		metadata["location"] = {"region": labels["region"]}

		if tf_zone.jurisdiction:
		metadata["jurisdiction"] = dict(tf_zone.jurisdiction)
		if tf_zone.legal_entity is not None:
		metadata["legal_entity"] = tf_zone.legal_entity
		if tf_zone.federates_zone_refs:
		metadata["federates_zone_refs"] = list(tf_zone.federates_zone_refs)

		return Zone(
		id=well_known_zone_id(tf_zone.ref),
		ref=tf_zone.ref,
		name=tf_zone.name,
		kind=tf_zone.kind,
		state=tf_zone.state,
		metadata=metadata,
		)


		def _domain_from(tf_domain: TransformationFunctionDomain, zone_id: UUID) -> Domain:
		metadata: dict[str, Any] = {}
		if tf_domain.details:
		metadata.update(tf_domain.details)
		if tf_domain.domain_type is not None:
		metadata["domain_type"] = tf_domain.domain_type
		if tf_domain.adapter_type is not None:
		metadata["adapter_type"] = tf_domain.adapter_type
		if tf_domain.federation_role is not None:
		metadata["federation_role"] = tf_domain.federation_role
		if tf_domain.federates_domain_refs:
		metadata["federates_domain_refs"] = list(tf_domain.federates_domain_refs)

		return Domain(
		zone_id=zone_id,
		ref=tf_domain.ref,
		name=tf_domain.name,
		kind=tf_domain.kind,
		state=tf_domain.state,
		metadata=metadata,
		)


		def _capability_from(
		tf_capability: TransformationFunctionCapability,
		domain_id: UUID,
		) -> Capability:
		return Capability(
		domain_id=domain_id,
		ref=tf_capability.ref,
		name=tf_capability.name,
		kind=tf_capability.kind,
		state=tf_capability.state,
		metadata=dict(tf_capability.metadata or {}),
		)


		def _binding_from(
		tf_binding: TransformationFunctionControlPathBinding,
		capability_id: UUID,
		) -> ControlPathBinding:
		return ControlPathBinding(
		capability_id=capability_id,
		ref=tf_binding.ref,
		control_path_ref=tf_binding.control_path_ref,
		supported_runtime_kinds=list(tf_binding.supported_runtime_kinds),
		supported_actions=list(tf_binding.supported_actions),
		priority=tf_binding.priority,
		is_default=tf_binding.is_default,
		state=tf_binding.state,
		metadata=dict(tf_binding.metadata or {}),
		)


		def _reject_duplicate(seen: set[str], ref: str, *, entity: str, scope: str) -> None:
		if ref in seen:
		raise BootstrapTopologyError(f"duplicate {entity} ref across {scope}: {ref!r}")
		seen.add(ref)


		async def sync_bootstrap_topology(
		discovery: TopologyDiscoveryPort,
		*,
		zones: ZoneRepository,
		domains: DomainRepository,
		capabilities: CapabilityRepository,
		bindings: ControlPathBindingRepository,
		known_control_path_refs: set[str],
		) -> None:
		"""Create missing bootstrap topology rows; never update or delete existing ones."""
		seen_domain_refs: set[str] = set()
		seen_capability_refs: set[str] = set()
		warned_control_path_refs: set[str] = set()

		for tf_zone in discovery.get_zones():
		existing_zone = await zones.get_by_ref(tf_zone.ref)
		if existing_zone is None:
		zone = await zones.create(_zone_from(tf_zone))
		logger.info("bootstrap_zone_created", ref=zone.ref, id=str(zone.id))
		else:
		zone = existing_zone
		logger.debug("bootstrap_zone_exists", ref=zone.ref, id=str(zone.id))

		for tf_domain in discovery.get_domains(zone_ref=tf_zone.ref):
		_reject_duplicate(seen_domain_refs, tf_domain.ref, entity="domain", scope="zones")
		existing_domain = await domains.get_by_ref(zone.id, tf_domain.ref)
		if existing_domain is None:
		domain = await domains.create(_domain_from(tf_domain, zone.id))
		logger.info("bootstrap_domain_created", ref=domain.ref, id=str(domain.id))
		else:
		domain = existing_domain
		logger.debug("bootstrap_domain_exists", ref=domain.ref, id=str(domain.id))

		for tf_capability in discovery.get_capabilities(domain_ref=tf_domain.ref):
		_reject_duplicate(
		seen_capability_refs,
		tf_capability.ref,
		entity="capability",
		scope="domains",
		)
		existing_capability = await capabilities.get_by_ref(domain.id, tf_capability.ref)
		if existing_capability is None:
		capability = await capabilities.create(
		_capability_from(tf_capability, domain.id)
		)
		logger.info(
		"bootstrap_capability_created",
		ref=capability.ref,
		id=str(capability.id),
		)
		else:
		capability = existing_capability
		logger.debug(
		"bootstrap_capability_exists",
		ref=capability.ref,
		id=str(capability.id),
		)

		for tf_binding in discovery.get_control_path_bindings(
		capability_ref=tf_capability.ref
		):
		control_path_ref = tf_binding.control_path_ref
		if (
		control_path_ref not in known_control_path_refs
		and control_path_ref not in warned_control_path_refs
		):
		warned_control_path_refs.add(control_path_ref)
		# TODO(ADR-0041 / architecture F5): fail startup once
		# ControlPathRegistry is built from the bootstrap YAML
		# control_paths section.
		logger.warning(
		"bootstrap_binding_unknown_control_path",
		binding_ref=tf_binding.ref,
		control_path_ref=control_path_ref,
		)

		existing_binding = await bindings.get_by_ref(capability.id, tf_binding.ref)
		if existing_binding is None:
		binding = await bindings.create(_binding_from(tf_binding, capability.id))
		logger.info(
		"bootstrap_binding_created",
		ref=binding.ref,
		id=str(binding.id),
		)
		else:
		logger.debug(
		"bootstrap_binding_exists",
		ref=existing_binding.ref,
		id=str(existing_binding.id),
		)

src/srm/domain/ports/database/topology.py

+16 −0

Changes for src/srm/domain/ports/database/topology.py: 16 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -17,6 +17,10 @@ class ZoneRepository(ABC):
		async def get_by_id(self, id: UUID) -> Zone \| None:
		pass

		@abstractmethod
		async def get_by_ref(self, ref: str) -> Zone \| None:
		pass

		@abstractmethod
		async def list_active_resource_zones(self) -> list[Zone]:
		pass
		@@ -40,6 +44,10 @@ class DomainRepository(ABC):
		async def get_by_id(self, id: UUID) -> Domain \| None:
		pass

		@abstractmethod
		async def get_by_ref(self, zone_id: UUID, ref: str) -> Domain \| None:
		pass

		@abstractmethod
		async def create(self, domain: Domain) -> Domain:
		pass
		@@ -50,6 +58,10 @@ class CapabilityRepository(ABC):
		async def get_by_id(self, id: UUID) -> Capability \| None:
		pass

		@abstractmethod
		async def get_by_ref(self, domain_id: UUID, ref: str) -> Capability \| None:
		pass

		@abstractmethod
		async def create(self, capability: Capability) -> Capability:
		pass
		@@ -60,6 +72,10 @@ class ControlPathBindingRepository(ABC):
		async def get_by_id(self, id: UUID) -> ControlPathBinding \| None:
		pass

		@abstractmethod
		async def get_by_ref(self, capability_id: UUID, ref: str) -> ControlPathBinding \| None:
		pass

		@abstractmethod
		async def create(self, control_path_binding: ControlPathBinding) -> ControlPathBinding:
		pass

src/srm/main.py

+28 −6

Changes for src/srm/main.py: 28 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -22,6 +22,12 @@ from srm.adapters.database.core import (
		schema_initialization,
		)
		from srm.adapters.database.repos.catalog import SqlServiceSpecificationRepository
		from srm.adapters.database.repos.topology import (
		SqlCapabilityRepository,
		SqlControlPathBindingRepository,
		SqlDomainRepository,
		SqlZoneRepository,
		)
		from srm.adapters.databus.nats_connection_manager import (
		NatsConnectionManager,
		init_databus_manager,
		@@ -43,6 +49,7 @@ from srm.api.rest.router import internal
		from srm.application.bootstrap.platform_catalog import (
		ensure_platform_service_specifications,
		)
		from srm.application.bootstrap.topology import sync_bootstrap_topology
		from srm.config import Settings, get_settings


		@@ -57,6 +64,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
		logger = structlog.get_logger()
		logger.info("Starting", app=settings.app_name, version=settings.app_version)

		control_path_registry = build_control_path_registry(
		settings.control_path_settings,
		factory=OpenOpTfSdkAdapterFactory(),
		)
		topology_discovery = build_topology_discovery(settings.bootstrap_topology_path)

		engine = None
		try:
		engine, session_maker = await build_engine_and_session_maker(
		url=settings.postgres_settings.url,
		@@ -73,6 +87,17 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
		platform_app_provider_id=settings.bootstrap_settings.platform_app_provider_id,
		overrides=settings.bootstrap_settings.service_specification_ids,
		)
		if topology_discovery is None:
		logger.info("bootstrap_topology_not_configured")
		else:
		await sync_bootstrap_topology(
		topology_discovery,
		zones=SqlZoneRepository(session),
		domains=SqlDomainRepository(session),
		capabilities=SqlCapabilityRepository(session),
		bindings=SqlControlPathBindingRepository(session),
		known_control_path_refs=set(control_path_registry.control_path_refs()),
		)
		await session.commit()
		except Exception:
		await session.rollback()
		@@ -80,6 +105,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:

		except Exception as e:
		logger.error("Database engine init failed!", error=str(e))
		if engine is not None:
		await engine.dispose()
		raise
		try:
		databus_manager: NatsConnectionManager = await init_databus_manager(
		@@ -95,11 +122,6 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
		settings=settings.federation_manager_settings, publisher=publisher
		)

		control_path_registry = build_control_path_registry(
		settings.control_path_settings,
		factory=OpenOpTfSdkAdapterFactory(),
		)

		try:
		databus_subscribers: list[
		NatsSubscriber \| NatsEventSubscriber
		@@ -117,7 +139,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
		raise

		app.state.control_path_registry = control_path_registry
		app.state.topology_discovery = build_topology_discovery(settings.bootstrap_topology_path)
		app.state.topology_discovery = topology_discovery
		app.state.db_engine = engine
		app.state.session_maker = session_maker
		app.state.databus_connection_manager = databus_manager