SRM Refactor: Service Specification Endpoints (!17) · Merge requests · OOP / code / service-resource-manager

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

+60 −2

Changes for src/srm/adapters/database/repos/catalog.py: 60 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,7 +3,7 @@ from __future__ import annotations
		from uuid import UUID

		import structlog
		from sqlalchemy import select
		from sqlalchemy import delete, select
		from sqlalchemy.exc import IntegrityError
		from sqlalchemy.ext.asyncio import AsyncSession
		from sqlalchemy.orm import selectinload
		@@ -16,14 +16,17 @@ from srm.adapters.database.mappers import (
		from srm.adapters.database.sql import (
		ServiceCapabilityRequirementRow,
		ServiceDeploymentUnitRow,
		ServiceInstanceRow,
		ServiceSpecificationRow,
		)
		from srm.adapters.errors import DuplicateServiceSpecificationError
		from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError
		from srm.domain.models.catalog import (
		ServiceCapabilityRequirement,
		ServiceDeploymentUnit,
		ServiceSpecification,
		)
		from srm.domain.models.catalog.enums import ServiceSpecificationState
		from srm.domain.models.runtime_inventory.enums import ServiceInstanceState
		from srm.domain.ports.database.catalog import (
		ServiceCapabilityRequirementRepository,
		ServiceDeploymentUnitRepository,
		@@ -47,6 +50,13 @@ def _log_integrity_error(exc: IntegrityError, *, entity: str, operation: str) ->
		)


		def _link_requirement_deployment_unit_rows(row: ServiceSpecificationRow) -> None:
		deployment_units_by_id = {unit.id: unit for unit in row.deployment_units}
		for requirement in row.capability_requirements:
		if requirement.deployment_unit_id in deployment_units_by_id:
		requirement.deployment_unit = deployment_units_by_id[requirement.deployment_unit_id]


		class SqlServiceSpecificationRepository(ServiceSpecificationRepository):
		def __init__(self, session: AsyncSession) -> None:
		self._session = session
		@@ -65,6 +75,7 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository):

		async def create(self, service_specification: ServiceSpecification) -> ServiceSpecification:
		row = ServiceSpecificationMapper.to_row(service_specification)
		_link_requirement_deployment_unit_rows(row)
		self._session.add(row)
		try:
		await self._session.flush()
		@@ -79,6 +90,53 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository):
		raise RuntimeError("Created service specification could not be reloaded")
		return saved

		async def delete_or_retire(self, id: UUID) -> bool:
		row = await self._session.get(ServiceSpecificationRow, id)
		if row is None:
		return False

		any_instance_id = await self._session.scalar(
		select(ServiceInstanceRow.id)
		.where(ServiceInstanceRow.service_specification_id == id)
		.limit(1)
		)
		if any_instance_id is None:
		await self._session.execute(
		delete(ServiceCapabilityRequirementRow).where(
		ServiceCapabilityRequirementRow.service_specification_id == id
		)
		)
		await self._session.execute(
		delete(ServiceDeploymentUnitRow).where(
		ServiceDeploymentUnitRow.service_specification_id == id
		)
		)
		await self._session.execute(
		delete(ServiceSpecificationRow).where(ServiceSpecificationRow.id == id)
		)
		await self._session.flush()
		return True

		live_instance_id = await self._session.scalar(
		select(ServiceInstanceRow.id)
		.where(ServiceInstanceRow.service_specification_id == id)
		.where(
		ServiceInstanceRow.state.not_in(
		[
		ServiceInstanceState.TERMINATED,
		ServiceInstanceState.FAILED,
		]
		)
		)
		.limit(1)
		)
		if live_instance_id is not None:
		raise ServiceSpecificationInUseError()

		row.state = ServiceSpecificationState.RETIRED
		await self._session.flush()
		return True


		class SqlServiceDeploymentUnitRepository(ServiceDeploymentUnitRepository):
		def __init__(self, session: AsyncSession) -> None:

src/srm/adapters/database/sql.py

+3 −0

Changes for src/srm/adapters/database/sql.py: 3 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -227,6 +227,9 @@ class ServiceCapabilityRequirementRow(AuditedMixin, Base):
		deployment_unit_id: Mapped[UUID \| None] = mapped_column(
		ForeignKey("service_deployment_unit.id")
		)
		deployment_unit: Mapped["ServiceDeploymentUnitRow \| None"] = relationship(
		foreign_keys=[deployment_unit_id]
		)
		ref: Mapped[str] = mapped_column(String(255), nullable=False)
		capability_kind: Mapped[CapabilityKind] = mapped_column(
		_enum_type(CapabilityKind), nullable=False

src/srm/adapters/errors.py

+5 −0

Changes for src/srm/adapters/errors.py: 5 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -10,3 +10,8 @@ class DuplicateEntryError(Exception):

		class DuplicateServiceSpecificationError(DuplicateEntryError):
		entity_name: ClassVar[str] = "Service Specification"


		class ServiceSpecificationInUseError(Exception):
		def __init__(self) -> None:
		super().__init__("Service specification has live service instances.")

src/srm/api/rest/router.py

0 → 100644

+238 −0

Changes for src/srm/api/rest/router.py: 238 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from uuid import UUID

		import structlog
		from fastapi import APIRouter, HTTPException, status

		from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError
		from srm.api.dependencies import (
		CreateServiceSpecificationUseCaseDep,
		DeleteServiceSpecificationUseCaseDep,
		GetServiceSpecificationUseCaseDep,
		)
		from srm.api.rest.schemas import (
		CreateServiceSpecificationRequest,
		CreateServiceSpecificationResponse,
		ErrorResponse,
		GetServiceSpecificationResponse,
		ServiceCapabilityRequirementResponseSchema,
		ServiceDeploymentUnitCreateSchema,
		ServiceSpecificationCreateSchema,
		)
		from srm.domain.models.canonical_parameters.parameters import (
		CapabilityParameters,
		CapabilityTarget,
		Parameters,
		SourceSpecification,
		SourceSpecificationFamily,
		)
		from srm.domain.models.catalog import (
		ServiceCapabilityRequirement,
		ServiceDeploymentUnit,
		ServiceSpecification,
		)

		internal = APIRouter(prefix="/internal", tags=["Internal"])
		logger: structlog.BoundLogger = structlog.get_logger(__name__)


		def _default_parameters(specification: ServiceSpecification) -> Parameters:
		source_api = str(specification.descriptor.get("source_api", "internal"))
		return Parameters(
		target=CapabilityTarget(),
		parameters=CapabilityParameters(),
		source_spec=SourceSpecification(
		family=SourceSpecificationFamily.INTERNAL,
		api=source_api,
		version=specification.version,
		),
		)


		def _to_domain(request: CreateServiceSpecificationRequest) -> ServiceSpecification:
		exclude_fields = set()
		if request.service_specification.id is None:
		exclude_fields.add("id")

		specification = ServiceSpecification.model_validate(
		request.service_specification.model_dump(mode="python", exclude=exclude_fields)
		)
		default_parameters = _default_parameters(specification)

		specification.deployment_units = [
		ServiceDeploymentUnit.model_validate(
		{
		**unit.model_dump(mode="python"),
		"service_specification_id": specification.id,
		}
		)
		for unit in request.service_deployment_units
		]
		specification.capability_requirements = [
		ServiceCapabilityRequirement.model_validate(
		{
		**requirement.model_dump(mode="python", exclude={"deployment_unit_ref"}),
		"service_specification_id": specification.id,
		"deployment_unit_id": None,
		"parameters": default_parameters.model_copy(deep=True).model_dump(mode="python"),
		}
		)
		for requirement in request.service_capability_requirements
		]

		return specification


		def _requirement_deployment_unit_refs(
		request: CreateServiceSpecificationRequest,
		) -> dict[str, str]:
		return {
		requirement.ref: requirement.deployment_unit_ref
		for requirement in request.service_capability_requirements
		if requirement.deployment_unit_ref is not None
		}


		def _build_response(saved: ServiceSpecification) -> CreateServiceSpecificationResponse:
		return CreateServiceSpecificationResponse(id=saved.id, ref=saved.ref)


		def _build_get_response(saved: ServiceSpecification) -> GetServiceSpecificationResponse:
		return GetServiceSpecificationResponse(
		service_specification=ServiceSpecificationCreateSchema(
		id=saved.id,
		app_provider_id=saved.app_provider_id,
		ref=saved.ref,
		name=saved.name,
		version=saved.version,
		descriptor=saved.descriptor,
		metadata=saved.metadata,
		),
		service_deployment_units=[
		ServiceDeploymentUnitCreateSchema(
		ref=unit.ref,
		name=unit.name,
		runtime_kind=unit.runtime_kind,
		artifact_ref=unit.artifact_ref,
		resource_requirements=unit.resource_requirements,
		parameters_schema=unit.parameters_schema,
		metadata=unit.metadata,
		)
		for unit in saved.deployment_units
		],
		service_capability_requirements=[
		ServiceCapabilityRequirementResponseSchema(
		ref=requirement.ref,
		capability_kind=requirement.capability_kind,
		domain_kind=requirement.domain_kind,
		is_required=requirement.is_required,
		selector=requirement.selector,
		policy=requirement.policy,
		metadata=requirement.metadata,
		)
		for requirement in saved.capability_requirements
		],
		)


		@internal.post(
		"/catalog/service-specifications",
		status_code=status.HTTP_201_CREATED,
		responses={
		status.HTTP_409_CONFLICT: {
		"model": ErrorResponse,
		"description": "Service specification already exists.",
		}
		},
		)
		async def create_service_specification(
		request: CreateServiceSpecificationRequest,
		use_case: CreateServiceSpecificationUseCaseDep,
		) -> CreateServiceSpecificationResponse:
		logger.info(
		"create_service_specification_requested",
		service_specification_ref=request.service_specification.ref,
		version=request.service_specification.version,
		)

		try:
		saved = await use_case.execute(
		_to_domain(request),
		_requirement_deployment_unit_refs(request),
		)
		except DuplicateServiceSpecificationError as exc:
		raise HTTPException(
		status_code=status.HTTP_409_CONFLICT,
		detail="Service specification already exists.",
		) from exc

		logger.info(
		"create_service_specification_succeeded",
		service_specification_id=str(saved.id),
		)
		return _build_response(saved)


		@internal.get(
		"/catalog/service-specifications/{id}",
		responses={
		status.HTTP_404_NOT_FOUND: {
		"model": ErrorResponse,
		"description": "Service specification not found.",
		}
		},
		)
		async def get_service_specification(
		id: UUID,
		use_case: GetServiceSpecificationUseCaseDep,
		) -> GetServiceSpecificationResponse:
		logger.info("get_service_specification_requested", service_specification_id=str(id))

		specification = await use_case.execute(id)
		if specification is None:
		raise HTTPException(
		status_code=status.HTTP_404_NOT_FOUND,
		detail="Service specification not found.",
		)

		logger.info(
		"get_service_specification_succeeded",
		service_specification_id=str(specification.id),
		)
		return _build_get_response(specification)


		@internal.delete(
		"/catalog/service-specifications/{id}",
		status_code=status.HTTP_204_NO_CONTENT,
		responses={
		status.HTTP_404_NOT_FOUND: {
		"model": ErrorResponse,
		"description": "Service specification not found.",
		},
		status.HTTP_409_CONFLICT: {
		"model": ErrorResponse,
		"description": "Service specification has live service instances.",
		},
		},
		)
		async def delete_service_specification(
		id: UUID,
		use_case: DeleteServiceSpecificationUseCaseDep,
		) -> None:
		logger.info("delete_service_specification_requested", service_specification_id=str(id))

		try:
		deleted = await use_case.execute(id)
		except ServiceSpecificationInUseError as exc:
		raise HTTPException(
		status_code=status.HTTP_409_CONFLICT,
		detail="Service specification has live service instances.",
		) from exc

		if not deleted:
		raise HTTPException(
		status_code=status.HTTP_404_NOT_FOUND,
		detail="Service specification not found.",
		)

		logger.info("delete_service_specification_succeeded", service_specification_id=str(id))

src/srm/api/rest/schemas.py

0 → 100644

+91 −0

Changes for src/srm/api/rest/schemas.py: 91 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from typing import Any
		from uuid import UUID

		from pydantic import Field, model_validator

		from srm.domain.models.canonical_parameters.compute import ComputeRequirements
		from srm.domain.models.catalog.enums import RuntimeKind
		from srm.domain.models.common import DomainModel
		from srm.domain.models.topology.enums import CapabilityKind, DomainKind


		class ServiceSpecificationCreateSchema(DomainModel):
		id: UUID \| None = None
		app_provider_id: str
		ref: str
		name: str
		version: str
		descriptor: dict[str, Any] = Field(default_factory=dict)
		metadata: dict[str, Any] = Field(default_factory=dict)


		class ServiceDeploymentUnitCreateSchema(DomainModel):
		ref: str
		name: str
		runtime_kind: RuntimeKind
		artifact_ref: str \| None = None
		resource_requirements: ComputeRequirements
		parameters_schema: dict[str, Any] = Field(default_factory=dict)
		metadata: dict[str, Any] = Field(default_factory=dict)


		class ServiceCapabilityRequirementCreateSchema(DomainModel):
		ref: str
		deployment_unit_ref: str \| None = None
		capability_kind: CapabilityKind
		domain_kind: DomainKind \| None = None
		is_required: bool = True
		selector: dict[str, Any] = Field(default_factory=dict)
		policy: dict[str, Any] = Field(default_factory=dict)
		metadata: dict[str, Any] = Field(default_factory=dict)


		class ServiceCapabilityRequirementResponseSchema(DomainModel):
		ref: str
		capability_kind: CapabilityKind
		domain_kind: DomainKind \| None = None
		is_required: bool = True
		selector: dict[str, Any] = Field(default_factory=dict)
		policy: dict[str, Any] = Field(default_factory=dict)
		metadata: dict[str, Any] = Field(default_factory=dict)


		class CreateServiceSpecificationRequest(DomainModel):
		service_specification: ServiceSpecificationCreateSchema
		service_deployment_units: list[ServiceDeploymentUnitCreateSchema] = Field(default_factory=list)
		service_capability_requirements: list[ServiceCapabilityRequirementCreateSchema] = Field(
		default_factory=list
		)

		@model_validator(mode="after")
		def validate_deployment_unit_refs(self) -> "CreateServiceSpecificationRequest":
		deployment_unit_refs = {unit.ref for unit in self.service_deployment_units}
		missing_refs = sorted(
		{
		requirement.deployment_unit_ref
		for requirement in self.service_capability_requirements
		if requirement.deployment_unit_ref is not None
		and requirement.deployment_unit_ref not in deployment_unit_refs
		}
		)
		if missing_refs:
		raise ValueError(
		"Unknown deployment unit refs in service_capability_requirements: "
		+ ", ".join(missing_refs)
		)
		return self


		class CreateServiceSpecificationResponse(DomainModel):
		id: UUID
		ref: str


		class GetServiceSpecificationResponse(DomainModel):
		service_specification: ServiceSpecificationCreateSchema
		service_deployment_units: list[ServiceDeploymentUnitCreateSchema]
		service_capability_requirements: list[ServiceCapabilityRequirementResponseSchema]


		class ErrorResponse(DomainModel):
		detail: str

SRM Refactor: Service Specification Endpoints

Merge request reports