feat: service specifications internal endpoints (b9c699c4) · Commits · OOP / code / service-resource-manager

src/srm/api/dependencies.py

+27 −0

Original line number	Diff line number	Diff line
		@@ -3,7 +3,12 @@ from typing import Annotated, AsyncGenerator, cast
		from fastapi import Depends, Request
		from sqlalchemy.ext.asyncio import AsyncSession

		from srm.adapters.database.repos.catalog import SqlServiceSpecificationRepository
		from srm.app_state import AppState
		from srm.application.use_cases.catalog import (
		CreateServiceSpecificationUseCase,
		GetServiceSpecificationUseCase,
		)


		def get_app_state(request: Request) -> AppState:
		@@ -21,3 +26,25 @@ async def get_session(request: Request) -> AsyncGenerator[AsyncSession, None]:


		SessionDep = Annotated[AsyncSession, Depends(get_session)]


		def get_create_service_specification_use_case(
		session: SessionDep,
		) -> CreateServiceSpecificationUseCase:
		return CreateServiceSpecificationUseCase(SqlServiceSpecificationRepository(session))


		def get_service_specification_use_case(
		session: SessionDep,
		) -> GetServiceSpecificationUseCase:
		return GetServiceSpecificationUseCase(SqlServiceSpecificationRepository(session))


		CreateServiceSpecificationUseCaseDep = Annotated[
		CreateServiceSpecificationUseCase,
		Depends(get_create_service_specification_use_case),
		]
		GetServiceSpecificationUseCaseDep = Annotated[
		GetServiceSpecificationUseCase,
		Depends(get_service_specification_use_case),
		]

src/srm/api/rest/router.py

0 → 100644

+196 −0

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
		from srm.api.dependencies import (
		CreateServiceSpecificationUseCaseDep,
		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:
		specification = ServiceSpecification.model_validate(
		request.service_specification.model_dump(mode="python")
		)
		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)


		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)

src/srm/api/rest/schemas.py

0 → 100644

+90 −0

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
		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


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


		class ErrorResponse(DomainModel):
		detail: str

src/srm/application/use_cases/catalog.py

0 → 100644

+38 −0

Original line number	Diff line number	Diff line
		from uuid import UUID

		from srm.domain.models.catalog import ServiceSpecification
		from srm.domain.ports.database.catalog import ServiceSpecificationRepository


		class CreateServiceSpecificationUseCase:
		def __init__(self, repository: ServiceSpecificationRepository) -> None:
		self._repository = repository

		async def execute(
		self,
		specification: ServiceSpecification,
		requirement_deployment_unit_refs: dict[str, str],
		) -> ServiceSpecification:
		saved = await self._repository.save(specification)

		deployment_unit_ids = {unit.ref: unit.id for unit in saved.deployment_units}
		for requirement in saved.capability_requirements:
		deployment_unit_ref = requirement_deployment_unit_refs.get(requirement.ref)
		if deployment_unit_ref is not None:
		requirement.deployment_unit_id = deployment_unit_ids[deployment_unit_ref]

		if any(
		requirement.deployment_unit_id is not None
		for requirement in saved.capability_requirements
		):
		saved = await self._repository.save(saved)

		return saved


		class GetServiceSpecificationUseCase:
		def __init__(self, repository: ServiceSpecificationRepository) -> None:
		self._repository = repository

		async def execute(self, id: UUID) -> ServiceSpecification \| None:
		return await self._repository.get_by_id(id)

src/srm/main.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -28,6 +28,7 @@ from srm.adapters.databus.nats_connection_manager import (
		from srm.api.databus.nats_subscriber import NatsSubscriber, subscribe_to_subjects
		from srm.api.health import health_router
		from srm.api.middlewares.middlewares import register_middlewares
		from srm.api.rest.router import internal
		from srm.config import Settings, get_settings


		@@ -101,5 +102,6 @@ def create_app() -> FastAPI:

		# Add routers
		app.include_router(health_router)
		app.include_router(internal)

		return app

Admin message