SRM Refactor: First version of TF-SDK Ports (!16) · Merge requests · OOP / code / service-resource-manager

src/srm/adapters/background_tasks/instances_monitor.py

0 → 100644

+67 −0

Changes for src/srm/adapters/background_tasks/instances_monitor.py: 67 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Implementation of Service and Capability Instances monitoring adapter
		using APScheduler package.
		"""

		import structlog
		from apscheduler.schedulers.asyncio import AsyncIOScheduler
		from apscheduler.triggers.interval import IntervalTrigger
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

		from srm.adapters.database.repos.runtime_inventory import (
		SqlCapabilityInstanceRepository,
		SqlServiceInstanceRepository,
		SqlServiceOrderRepository,
		)
		from srm.application.use_cases.instance_monitor import CapabilityInstanceMonitorUseCase
		from srm.domain.ports.background_tasks.instances_monitor import InstancesMonitorPort
		from srm.domain.ports.databus.publisher import DataBusPublisher
		from srm.domain.ports.transformation_functions import ControlPathRegistry

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


		class APSchedulerInstancesMonitor(InstancesMonitorPort):
		def __init__(
		self,
		session_maker: async_sessionmaker[AsyncSession],
		control_path_registry: ControlPathRegistry,
		publisher: DataBusPublisher,
		) -> None:
		self._session_maker = session_maker
		self._control_path_registry = control_path_registry
		self._publisher = publisher
		self._scheduler = AsyncIOScheduler()

		def register_scheduler(
		self,
		interval: float,
		) -> None:
		self._scheduler.add_job(
		self.update_statuses, IntervalTrigger(seconds=interval), id="instance_monitor"
		)

		def start(self) -> None:
		self._scheduler.start()

		def shutdown(self) -> None:
		self._scheduler.shutdown(wait=True)

		async def update_statuses(self) -> None:
		async with self._session_maker() as session:
		try:
		use_case = CapabilityInstanceMonitorUseCase(
		control_path_registry=self._control_path_registry,
		capability_instances=SqlCapabilityInstanceRepository(session=session),
		service_instances=SqlServiceInstanceRepository(session=session),
		service_orders=SqlServiceOrderRepository(session=session),
		publisher=self._publisher,
		)
		await use_case.execute()
		await session.commit()
		except Exception as exc:
		logger.exception(
		"instances_monitor_tick_failed",
		error=str(exc),
		error_type=type(exc).__name__,
		)
		await session.rollback()

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

+50 −0

Changes for src/srm/adapters/database/repos/catalog.py: 50 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -17,6 +17,7 @@ from srm.adapters.database.sql import (
		ServiceCapabilityRequirementRow,
		ServiceDeploymentUnitRow,
		ServiceInstanceRow,
		ServiceOrderRow,
		ServiceSpecificationRow,
		)
		from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError
		@@ -73,6 +74,24 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository):
		row = await self._session.scalar(stmt)
		return ServiceSpecificationMapper.to_domain(row) if row is not None else None

		async def list_by_app_provider_id(self, app_provider_id: str) -> list[ServiceSpecification]:
		stmt = (
		select(ServiceSpecificationRow)
		.options(
		selectinload(ServiceSpecificationRow.deployment_units),
		selectinload(ServiceSpecificationRow.capability_requirements),
		)
		.where(ServiceSpecificationRow.app_provider_id == app_provider_id)
		.order_by(
		ServiceSpecificationRow.name,
		ServiceSpecificationRow.ref,
		ServiceSpecificationRow.version,
		ServiceSpecificationRow.id,
		)
		)
		rows = (await self._session.scalars(stmt)).all()
		return [ServiceSpecificationMapper.to_domain(row) for row in rows]

		async def create(self, service_specification: ServiceSpecification) -> ServiceSpecification:
		row = ServiceSpecificationMapper.to_row(service_specification)
		_link_requirement_deployment_unit_rows(row)
		@@ -101,6 +120,17 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository):
		.limit(1)
		)
		if any_instance_id is None:
		# An instance a partner hosts leaves no instance row here, only its order, and that
		# order still points at the specification: keep it as retired history instead.
		any_order_id = await self._session.scalar(
		select(ServiceOrderRow.id)
		.where(ServiceOrderRow.service_specification_id == id)
		.limit(1)
		)
		if any_order_id is not None:
		row.state = ServiceSpecificationState.RETIRED
		await self._session.flush()
		return True
		await self._session.execute(
		delete(ServiceCapabilityRequirementRow).where(
		ServiceCapabilityRequirementRow.service_specification_id == id
		@@ -147,6 +177,16 @@ class SqlServiceDeploymentUnitRepository(ServiceDeploymentUnitRepository):
		row = await self._session.scalar(stmt)
		return ServiceDeploymentUnitMapper.to_domain(row) if row is not None else None

		async def list_by_service_specification_id(
		self,
		service_specification_id: UUID,
		) -> list[ServiceDeploymentUnit]:
		stmt = select(ServiceDeploymentUnitRow).where(
		ServiceDeploymentUnitRow.service_specification_id == service_specification_id
		)
		rows = (await self._session.scalars(stmt)).all()
		return [ServiceDeploymentUnitMapper.to_domain(row) for row in rows]

		async def create(self, deployment_unit: ServiceDeploymentUnit) -> ServiceDeploymentUnit:
		row = ServiceDeploymentUnitMapper.to_row(deployment_unit)
		self._session.add(row)
		@@ -173,6 +213,16 @@ class SqlServiceCapabilityRequirementRepository(ServiceCapabilityRequirementRepo
		row = await self._session.scalar(stmt)
		return ServiceCapabilityRequirementMapper.to_domain(row) if row is not None else None

		async def list_by_service_specification_id(
		self,
		service_specification_id: UUID,
		) -> list[ServiceCapabilityRequirement]:
		stmt = select(ServiceCapabilityRequirementRow).where(
		ServiceCapabilityRequirementRow.service_specification_id == service_specification_id
		)
		rows = (await self._session.scalars(stmt)).all()
		return [ServiceCapabilityRequirementMapper.to_domain(row) for row in rows]

		async def create(
		self,
		capability_requirement: ServiceCapabilityRequirement,

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

+295 −1

Changes for src/srm/adapters/database/repos/runtime_inventory.py: 295 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from datetime import datetime
		from uuid import UUID

		import structlog
		@@ -13,14 +14,31 @@ from srm.adapters.database.mappers import (
		ServiceInstanceMapper,
		ServiceOrderMapper,
		)
		from srm.adapters.database.sql import CapabilityInstanceRow, ServiceInstanceRow, ServiceOrderRow
		from srm.adapters.database.sql import (
		CapabilityInstanceRow,
		CapabilityRow,
		ControlPathBindingRow,
		DomainRow,
		ServiceInstanceRow,
		ServiceOrderRow,
		ZoneRow,
		)
		from srm.adapters.errors import DuplicateServiceInstanceError
		from srm.domain.models.canonical_parameters.result import Result
		from srm.domain.models.runtime_inventory import (
		CapabilityInstance,
		ServiceInstance,
		ServiceOrder,
		)
		from srm.domain.models.runtime_inventory.enums import (
		CapabilityInstanceState,
		ServiceInstanceState,
		ServiceOrderState,
		)
		from srm.domain.models.topology.enums import CapabilityKind
		from srm.domain.ports.database.runtime_inventory import (
		CapabilityInstanceRepository,
		MonitorableCapabilityInstance,
		ServiceInstanceRepository,
		ServiceOrderRepository,
		)
		@@ -51,6 +69,11 @@ class SqlServiceOrderRepository(ServiceOrderRepository):
		row = await self._session.scalar(stmt)
		return ServiceOrderMapper.to_domain(row) if row is not None else None

		async def get_by_operation_id(self, operation_id: UUID) -> ServiceOrder \| None:
		stmt = select(ServiceOrderRow).where(ServiceOrderRow.operation_id == operation_id)
		row = await self._session.scalar(stmt)
		return ServiceOrderMapper.to_domain(row) if row is not None else None

		async def create(self, service_order: ServiceOrder) -> ServiceOrder:
		row = ServiceOrderMapper.to_row(service_order)
		self._session.add(row)
		@@ -65,6 +88,41 @@ class SqlServiceOrderRepository(ServiceOrderRepository):
		raise RuntimeError("Created service order could not be reloaded")
		return saved

		async def update_state(
		self,
		id: UUID,
		state: ServiceOrderState,
		*,
		completed_at: datetime \| None = None,
		failure_detail: dict[str, object] \| None = None,
		) -> ServiceOrder:
		row = await self._session.get(ServiceOrderRow, id)
		if row is None:
		raise RuntimeError("Service order could not be reloaded")
		row.state = state
		row.completed_at = completed_at
		row.failure_detail = failure_detail
		await self._session.flush()
		saved = await self.get_by_id(id)
		if saved is None:
		raise RuntimeError("Updated service order could not be reloaded")
		return saved

		async def update_federated_handoff(
		self,
		id: UUID,
		federated_handoff: dict[str, object],
		) -> ServiceOrder:
		row = await self._session.get(ServiceOrderRow, id)
		if row is None:
		raise RuntimeError("Service order could not be reloaded")
		row.federated_handoff = federated_handoff
		await self._session.flush()
		saved = await self.get_by_id(id)
		if saved is None:
		raise RuntimeError("Updated service order could not be reloaded")
		return saved


		class SqlServiceInstanceRepository(ServiceInstanceRepository):
		def __init__(self, session: AsyncSession) -> None:
		@@ -81,6 +139,43 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository):
		row = await self._session.scalar(stmt)
		return ServiceInstanceMapper.to_domain(row) if row is not None else None

		async def list_by_app_provider_id(
		self,
		app_provider_id: str,
		service_instance_id: UUID \| None = None,
		service_specification_id: UUID \| None = None,
		) -> list[ServiceInstance]:
		stmt = (
		select(ServiceInstanceRow)
		.options(
		selectinload(ServiceInstanceRow.capability_instances),
		)
		.where(ServiceInstanceRow.app_provider_id == app_provider_id)
		)
		if service_instance_id is not None:
		stmt = stmt.where(ServiceInstanceRow.id == service_instance_id)
		if service_specification_id is not None:
		stmt = stmt.where(
		ServiceInstanceRow.service_specification_id == service_specification_id
		)

		rows = (await self._session.scalars(stmt)).all()
		return [ServiceInstanceMapper.to_domain(row) for row in rows]

		async def list_by_originating_service_order_id(
		self,
		originating_service_order_id: UUID,
		) -> list[ServiceInstance]:
		stmt = (
		select(ServiceInstanceRow)
		.options(
		selectinload(ServiceInstanceRow.capability_instances),
		)
		.where(ServiceInstanceRow.originating_service_order_id == originating_service_order_id)
		)
		rows = (await self._session.scalars(stmt)).all()
		return [ServiceInstanceMapper.to_domain(row) for row in rows]

		async def create(self, service_instance: ServiceInstance) -> ServiceInstance:
		row = ServiceInstanceMapper.to_row(service_instance)
		self._session.add(row)
		@@ -88,6 +183,8 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository):
		await self._session.flush()
		except IntegrityError as exc:
		_log_integrity_error(exc, entity="service_instance", operation="create")
		if _pgcode(exc) == "23505":
		raise DuplicateServiceInstanceError() from exc
		raise
		service_instance_id = row.id
		saved = await self.get_by_id(service_instance_id)
		@@ -95,6 +192,53 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository):
		raise RuntimeError("Created service instance could not be reloaded")
		return saved

		async def update_state(
		self,
		id: UUID,
		state: ServiceInstanceState,
		*,
		terminated_at: datetime \| None = None,
		) -> ServiceInstance:
		row = await self._session.get(ServiceInstanceRow, id)
		if row is None:
		raise RuntimeError("Service instance could not be reloaded")
		row.state = state
		if terminated_at is not None:
		row.terminated_at = terminated_at
		await self._session.flush()
		saved = await self.get_by_id(id)
		if saved is None:
		raise RuntimeError("Updated service instance could not be reloaded")
		return saved

		async def reconcile_states(self) -> int:
		stmt = (
		select(ServiceInstanceRow)
		.options(selectinload(ServiceInstanceRow.capability_instances))
		.where(ServiceInstanceRow.state != ServiceInstanceState.TERMINATED)
		)

		rows = (await self._session.scalars(stmt)).all()
		updated = 0
		for row in rows:
		previous_state = row.state
		service_instance = ServiceInstanceMapper.to_domain(row)
		service_instance.update_state()
		if service_instance.state == previous_state:
		continue
		row.state = service_instance.state
		row.terminated_at = service_instance.terminated_at
		updated += 1
		logger.info(
		"service_instance_state_changed",
		service_instance_id=str(row.id),
		from_state=previous_state,
		to_state=service_instance.state,
		)

		await self._session.flush()
		return updated


		class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		def __init__(self, session: AsyncSession) -> None:
		@@ -118,3 +262,153 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		if saved is None:
		raise RuntimeError("Created capability instance could not be reloaded")
		return saved

		async def update_state(
		self,
		id: UUID,
		state: CapabilityInstanceState,
		*,
		terminated_at: datetime \| None = None,
		result_summary: Result \| None = None,
		) -> CapabilityInstance:
		row = await self._session.get(CapabilityInstanceRow, id)
		if row is None:
		raise RuntimeError("Capability instance could not be reloaded")
		row.state = state
		if terminated_at is not None:
		row.terminated_at = terminated_at
		if result_summary is not None:
		row.result_summary = result_summary.model_dump(mode="json")
		await self._session.flush()
		saved = await self.get_by_id(id)
		if saved is None:
		raise RuntimeError("Updated capability instance could not be reloaded")
		return saved

		async def get_by_service_instance_id(
		self,
		service_instance_id: UUID \| None = None,
		) -> list[tuple[CapabilityInstance, CapabilityKind, UUID]]:
		logger.info(
		"get_by_service_instance_id",
		service_instance_id=service_instance_id,
		)
		stmt = (
		select(CapabilityInstanceRow, CapabilityRow.kind, ServiceInstanceRow.zone_id)
		.join(CapabilityRow)
		.where(CapabilityInstanceRow.capability_id == CapabilityRow.id)
		.join(ServiceInstanceRow)
		.where(ServiceInstanceRow.id == service_instance_id)
		.where(CapabilityInstanceRow.service_instance_id == service_instance_id)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.FAILED)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.TERMINATED)
		)

		rows = (await self._session.execute(stmt)).all()
		return [(CapabilityInstanceMapper.to_domain(row[0]), row[1], row[2]) for row in rows]

		async def list_by_app_provider_id(
		self,
		app_provider_id: str,
		capability_kind: CapabilityKind \| None = None,
		service_instance_id: UUID \| None = None,
		service_specification_id: UUID \| None = None,
		external_ref: str \| None = None,
		) -> list[tuple[CapabilityInstance, CapabilityKind, UUID]]:
		logger.info(
		"list_by_app_provider_id",
		app_provider_id=app_provider_id,
		capability_kind=capability_kind,
		service_instance_id=service_instance_id,
		service_specification_id=service_specification_id,
		external_ref=external_ref,
		)
		stmt = (
		select(CapabilityInstanceRow, CapabilityRow.kind, ServiceInstanceRow.zone_id)
		.join(CapabilityRow)
		.where(CapabilityInstanceRow.capability_id == CapabilityRow.id)
		.join(ServiceInstanceRow)
		.where(CapabilityInstanceRow.service_instance_id == ServiceInstanceRow.id)
		.where(CapabilityInstanceRow.app_provider_id == app_provider_id)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.FAILED)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.TERMINATED)
		)
		if capability_kind is not None:
		stmt = stmt.where(CapabilityRow.kind == capability_kind)
		if service_instance_id is not None:
		stmt = stmt.where(CapabilityInstanceRow.service_instance_id == service_instance_id)
		if service_specification_id is not None:
		stmt = stmt.where(
		ServiceInstanceRow.service_specification_id == service_specification_id
		)
		if external_ref is not None:
		stmt = stmt.where(CapabilityInstanceRow.external_ref == external_ref)
		rows = (await self._session.execute(stmt)).all()
		return [(CapabilityInstanceMapper.to_domain(row[0]), row[1], row[2]) for row in rows]

		async def list_monitorable(self) -> list[MonitorableCapabilityInstance]:
		stmt = (
		select(
		CapabilityInstanceRow,
		CapabilityRow.kind,
		CapabilityRow.ref,
		DomainRow.ref,
		DomainRow.kind,
		ZoneRow.ref,
		ZoneRow.kind,
		ControlPathBindingRow.ref,
		)
		.join(CapabilityRow, CapabilityInstanceRow.capability_id == CapabilityRow.id)
		.join(DomainRow, CapabilityRow.domain_id == DomainRow.id)
		.join(ZoneRow, DomainRow.zone_id == ZoneRow.id)
		.outerjoin(
		ControlPathBindingRow,
		CapabilityInstanceRow.control_path_binding_id == ControlPathBindingRow.id,
		)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.TERMINATED)
		.where(CapabilityInstanceRow.external_id.is_not(None))
		.where(CapabilityInstanceRow.control_path_ref_snapshot.is_not(None))
		)
		rows = (await self._session.execute(stmt)).all()
		monitorable: list[MonitorableCapabilityInstance] = []
		for (
		instance_row,
		capability_kind,
		capability_ref,
		domain_ref,
		domain_kind,
		zone_ref,
		zone_kind,
		binding_ref,
		) in rows:
		if instance_row.external_id is None or instance_row.control_path_ref_snapshot is None:
		continue
		monitorable.append(
		MonitorableCapabilityInstance(
		capability_instance_id=instance_row.id,
		service_instance_id=instance_row.service_instance_id,
		originating_service_order_id=instance_row.originating_service_order_id,
		external_id=instance_row.external_id,
		external_ref=instance_row.external_ref,
		control_path_ref=instance_row.control_path_ref_snapshot,
		control_path_binding_ref=binding_ref,
		capability_kind=capability_kind,
		capability_ref=capability_ref,
		domain_ref=domain_ref,
		domain_kind=domain_kind,
		zone_ref=zone_ref,
		zone_kind=zone_kind,
		current_state=instance_row.state,
		created_at=instance_row.created_at,
		duration_seconds=_duration_seconds(instance_row.parameters_snapshot),
		)
		)
		return monitorable


		def _duration_seconds(parameters_snapshot: dict[str, object]) -> int \| None:
		parameters = parameters_snapshot.get("parameters")
		if not isinstance(parameters, dict):
		return None
		value = parameters.get("duration_seconds")
		return value if isinstance(value, int) else None

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

+68 −15

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

		from typing import Any
		from uuid import UUID

		import structlog
		@@ -16,6 +17,7 @@ from srm.adapters.database.mappers import (
		)
		from srm.adapters.database.sql import CapabilityRow, ControlPathBindingRow, DomainRow, ZoneRow
		from srm.domain.models.topology import Capability, ControlPathBinding, Domain, Zone
		from srm.domain.models.topology.enums import ZoneKind, ZoneState
		from srm.domain.ports.database.topology import (
		CapabilityRepository,
		ControlPathBindingRepository,
		@@ -40,22 +42,59 @@ 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(_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

		async def list_active_resource_zones(self) -> list[Zone]:
		return await self.list_resource_zones(state=ZoneState.ACTIVE)

		async def list_resource_zones(
		self,
		*,
		state: ZoneState \| None = None,
		region: str \| None = None,
		) -> 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)
		)
		.where(ZoneRow.id == id)
		if state is not None:
		stmt = stmt.where(ZoneRow.state == state)
		if region is not None:
		stmt = stmt.where(
		ZoneRow.zone_metadata["location"]["region"].astext == region,
		)
		row = await self._session.scalar(stmt)
		return ZoneMapper.to_domain(row) if row is not None else None
		rows = (await self._session.scalars(stmt)).all()
		return [ZoneMapper.to_domain(row) for row in rows]

		async def create(self, zone: Zone) -> Zone:
		row = ZoneMapper.to_row(zone)
		@@ -77,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
		@@ -109,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
		@@ -141,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/database/mappers.py

+5 −3

Changes for src/srm/adapters/database/mappers.py: 5 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -295,7 +295,7 @@ class ServiceCapabilityRequirementMapper:
		domain_kind=domain.domain_kind,
		is_required=domain.is_required,
		selector=domain.selector,
		parameters=domain.parameters.model_dump(mode="python"),
		parameters=domain.parameters.model_dump(mode="json"),
		policy=domain.policy,
		capability_requirement_metadata=domain.metadata,
		created_at=domain.created_at,
		@@ -318,6 +318,7 @@ class ServiceOrderMapper:
		target_service_instance_id=row.target_service_instance_id,
		state=row.state,
		payload_snapshot=row.payload_snapshot,
		federated_handoff=row.federated_handoff,
		failure_detail=row.failure_detail,
		app_provider_id=row.app_provider_id,
		federation_partner_ref=row.federation_partner_ref,
		@@ -340,6 +341,7 @@ class ServiceOrderMapper:
		target_service_instance_id=domain.target_service_instance_id,
		state=domain.state,
		payload_snapshot=domain.payload_snapshot,
		federated_handoff=domain.federated_handoff,
		failure_detail=domain.failure_detail,
		app_provider_id=domain.app_provider_id,
		federation_partner_ref=domain.federation_partner_ref,
		@@ -436,8 +438,8 @@ class CapabilityInstanceMapper:
		state=domain.state,
		external_id=domain.external_id,
		external_ref=domain.external_ref,
		parameters_snapshot=domain.parameters_snapshot.model_dump(mode="python"),
		result_summary=domain.result_summary.model_dump(mode="python"),
		parameters_snapshot=domain.parameters_snapshot.model_dump(mode="json"),
		result_summary=domain.result_summary.model_dump(mode="json"),
		failure_detail=domain.failure_detail,
		capability_instance_metadata=domain.metadata,
		app_provider_id=domain.app_provider_id,

SRM Refactor: First version of TF-SDK Ports

Merge request reports