feat: refactor NATS subscriber and publisher, add tests for connection manager and publisher (b8a20797) · Commits · OOP / code / service-resource-manager

.env.example

+1 −1

Original line number	Diff line number	Diff line
		@@ -8,4 +8,4 @@ POSTGRES_SETTINGS__CREATE_SCHEMA_ON_STARTUP = true

		NATS_SETTINGS__URL = "nats://localhost:4222"
		NATS_SETTINGS__CONNECT_TIMEOUT = 10
		NATS_SETTINGS__MAX_RECONNECT_ATTEMPS = 3
		NATS_SETTINGS__MAX_RECONNECT_ATTEMPTS = 3

src/srm/api/databus/message_router.py

deleted100644 → 0

+0 −22

Original line number	Diff line number	Diff line
		from collections.abc import Awaitable, Callable

		from nats.aio.msg import Msg


		class MessageRouter:
		def __init__(self) -> None:
		self._handlers: dict[str, Callable[[Msg], Awaitable[None]]] = {}

		def register_handler(
		self,
		subject: str,
		handler: Callable[[Msg], Awaitable[None]],
		) -> None:
		self._handlers[subject] = handler

		async def route(self, msg: Msg) -> None:
		handler = self._handlers.get(msg.subject)
		if handler is None:
		raise RuntimeError(f"No handler registered for subject: {msg.subject}")

		await handler(msg)

src/srm/api/databus/nats_subscriber.py

+18 −44

Original line number	Diff line number	Diff line
		@@ -9,8 +9,18 @@ from srm.api.databus.schemas import InboundMessage

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

		COMMAND_SUBJECTS = [
		"command.srm.service.deploy",
		"command.srm.service.scale",
		"command.srm.service.terminate",
		"command.srm.network.capability.activate",
		"command.srm.network.capability.update",
		"command.srm.network.capability.deactivate",
		]


		async def _noop_router(message: InboundMessage) -> None:
		# TODO: the real router must validate in two stages (interface-contract.md §A, ADR-0032).
		logger.info("Message Received", subject=message.subject, data=message.payload)
		return None

		@@ -18,49 +28,10 @@ async def _noop_router(message: InboundMessage) -> None:
		async def subscribe_to_subjects(
		connection_manager: NatsConnectionManager,
		) -> list["NatsSubscriber"]:
		subscribers: list[NatsSubscriber] = []

		service_deploy_sub = NatsSubscriber(
		connection_manager=connection_manager,
		subject="command.srm.service.deploy",
		router=_noop_router,
		)
		subscribers.append(service_deploy_sub)

		service_scale_sub = NatsSubscriber(
		connection_manager=connection_manager,
		subject="command.srm.service.scale",
		router=_noop_router,
		)
		subscribers.append(service_scale_sub)

		service_terminate_sub = NatsSubscriber(
		connection_manager=connection_manager,
		subject="command.srm.service.terminate",
		router=_noop_router,
		)
		subscribers.append(service_terminate_sub)

		service_network_activate_sub = NatsSubscriber(
		connection_manager=connection_manager,
		subject="command.srm.network.capability.activate",
		router=_noop_router,
		)
		subscribers.append(service_network_activate_sub)

		service_network_update_sub = NatsSubscriber(
		connection_manager=connection_manager,
		subject="command.srm.network.capability.update",
		router=_noop_router,
		)
		subscribers.append(service_network_update_sub)

		service_network_deactivate_sub = NatsSubscriber(
		connection_manager=connection_manager,
		subject="command.srm.network.capability.deactivate",
		router=_noop_router,
		)
		subscribers.append(service_network_deactivate_sub)
		subscribers = [
		NatsSubscriber(connection_manager=connection_manager, subject=subject, router=_noop_router)
		for subject in COMMAND_SUBJECTS
		]

		for subscriber in subscribers:
		await subscriber.start()
		@@ -92,4 +63,7 @@ class NatsSubscriber:
		payload=msg.data,
		headers=dict(msg.headers) if msg.headers is not None else {},
		)
		try:
		await self._router(inbound_message)
		except Exception:
		logger.exception("databus_handler_failed", subject=msg.subject)

src/srm/api/databus/schemas.py

+39 −19

Original line number	Diff line number	Diff line
		@@ -38,21 +38,29 @@ class DeployPayloadV1(BaseModel):
		class DeployTargetV1(BaseModel):
		app_instance_id: UUID
		zone_id: UUID \| None = None
		domain_id: UUID \| None = None

		@model_validator(mode="after")
		def validate_pin_shape(self) -> "DeployTargetV1":
		if self.domain_id is not None and self.zone_id is None:
		raise ValueError("domain_id requires zone_id: a domain pin must name its zone")

		return self


		class SrmServiceDeployV1(CommandEnvelopeV1):
		service_specification_public_id: UUID
		targets: list[DeployTargetV1]
		service_specification_id: UUID
		targets: list[DeployTargetV1] = Field(min_length=1)
		deploy: DeployPayloadV1


		class ScalePayloadV1(BaseModel):
		replicas: int \| None = Field(default=None, ge=0)
		replicas: int = Field(ge=0)


		class SrmServiceScaleV1(CommandEnvelopeV1):
		service_instance_public_id: UUID
		service_specification_public_id: UUID \| None = None
		service_instance_id: UUID
		service_specification_id: UUID \| None = None
		scale: ScalePayloadV1


		@@ -61,8 +69,8 @@ class TerminatePayloadV1(BaseModel):


		class SrmServiceTerminateV1(CommandEnvelopeV1):
		service_instance_public_id: UUID
		service_specification_public_id: UUID \| None = None
		service_instance_id: UUID
		service_specification_id: UUID \| None = None
		terminate: TerminatePayloadV1


		@@ -74,35 +82,46 @@ class NetworkCapabilityPayloadV1(BaseModel):


		class SrmNetworkCapabilityActivateV1(CommandEnvelopeV1):
		service_specification_public_id: UUID
		zone_public_id: UUID \| None = None
		service_specification_id: UUID
		zone_id: UUID \| None = None
		domain_id: UUID \| None = None
		network_capability: NetworkCapabilityPayloadV1

		@model_validator(mode="after")
		def validate_pin_shape(self) -> "SrmNetworkCapabilityActivateV1":
		if self.domain_id is not None and self.zone_id is None:
		raise ValueError("domain_id requires zone_id: a domain pin must name its zone")

		return self


		class NetworkCapabilityRealizationRefV1(BaseModel):
		capability_type: str
		external_ref: str \| None = None
		service_instance_public_id: UUID \| None = None
		capability_type: str \| None = None
		service_instance_id: UUID \| None = None

		@model_validator(mode="after")
		def validate_reference_shape(self) -> "NetworkCapabilityRealizationRefV1":
		if self.external_ref is not None:
		return self

		if self.service_instance_public_id is not None and self.capability_type is not None:
		return self
		has_external_ref = self.external_ref is not None
		has_service_instance_id = self.service_instance_id is not None

		if has_external_ref == has_service_instance_id:
		raise ValueError(
		"network capability realization must be identified by external_ref "
		"or by service_instance_public_id and capability_type"
		"network capability realization must be identified by exactly one of "
		"external_ref or service_instance_id"
		)

		return self


		class NetworkCapabilityUpdatePayloadV1(NetworkCapabilityRealizationRefV1):
		target: CapabilityTarget \| None = None
		profile_ref: str \| None = None
		parameters: CapabilityParameters


		class SrmNetworkCapabilityUpdateV1(CommandEnvelopeV1):
		service_specification_id: UUID \| None = None
		network_capability: NetworkCapabilityUpdatePayloadV1


		@@ -111,4 +130,5 @@ class NetworkCapabilityDeactivatePayloadV1(NetworkCapabilityRealizationRefV1):


		class SrmNetworkCapabilityDeactivateV1(CommandEnvelopeV1):
		service_specification_id: UUID \| None = None
		network_capability: NetworkCapabilityDeactivatePayloadV1

tests/api/databus/init.py

0 → 100644

+0 −0

Empty file added.

Admin message