feat: add AppDeploymentRepository port and SQL adapter (22caa48b) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/database/mappers.py

+30 −0

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

		from open_exposure_gateway.adapters.database.sql import (
		AppDeploymentRow,
		AppInstanceRow,
		AppRegistrationRow,
		CallbackDeliveryRow,
		@@ -7,6 +10,7 @@ from open_exposure_gateway.adapters.database.sql import (
		QodSessionRow,
		)
		from open_exposure_gateway.domain.models import (
		AppDeployment,
		AppInstance,
		AppRegistration,
		CallbackDelivery,
		@@ -110,6 +114,32 @@ class AppInstanceMapper:
		)


		class AppDeploymentMapper:
		@staticmethod
		def to_domain(row: AppDeploymentRow) -> AppDeployment:
		return AppDeployment(
		app_deployment_id=row.app_deployment_id,
		operation_id=row.operation_id,
		app_registration_id=row.app_registration_id,
		app_deployment_name=row.app_deployment_name,
		edge_cloud_zones=[UUID(zone_id) for zone_id in row.edge_cloud_zones],
		state=row.state,
		created_at=row.created_at,
		updated_at=row.updated_at,
		)

		@staticmethod
		def to_row(domain: AppDeployment) -> AppDeploymentRow:
		return AppDeploymentRow(
		app_deployment_id=domain.app_deployment_id,
		operation_id=domain.operation_id,
		app_registration_id=domain.app_registration_id,
		app_deployment_name=domain.app_deployment_name,
		edge_cloud_zones=[str(zone_id) for zone_id in domain.edge_cloud_zones],
		state=domain.state,
		)


		class QodSessionMapper:
		@staticmethod
		def to_domain(row: QodSessionRow) -> QodSession:

src/open_exposure_gateway/adapters/database/repos/deployments.py

0 → 100644

+34 −0

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

		from sqlalchemy import select
		from sqlalchemy.ext.asyncio import AsyncSession

		from open_exposure_gateway.adapters.database.mappers import AppDeploymentMapper
		from open_exposure_gateway.adapters.database.sql import AppDeploymentRow
		from open_exposure_gateway.domain.models import AppDeployment
		from open_exposure_gateway.ports.database.deployments import AppDeploymentRepository


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

		async def get_by_id(self, app_deployment_id: UUID) -> AppDeployment \| None:
		stmt = select(AppDeploymentRow).where(
		AppDeploymentRow.app_deployment_id == app_deployment_id
		)
		row = await self._session.scalar(stmt)
		return AppDeploymentMapper.to_domain(row) if row is not None else None

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

		async def save(self, app_deployment: AppDeployment) -> AppDeployment:
		merged = await self._session.merge(AppDeploymentMapper.to_row(app_deployment))
		await self._session.flush()
		saved = await self.get_by_id(merged.app_deployment_id)
		if saved is None:
		raise RuntimeError("Saved app deployment could not be reloaded")
		return saved

src/open_exposure_gateway/ports/database/deployments.py

0 → 100644

+20 −0

Original line number	Diff line number	Diff line
		"""App deployment repository ports."""

		from abc import ABC, abstractmethod
		from uuid import UUID

		from open_exposure_gateway.domain.models import AppDeployment


		class AppDeploymentRepository(ABC):
		@abstractmethod
		async def get_by_id(self, app_deployment_id: UUID) -> AppDeployment \| None:
		pass

		@abstractmethod
		async def get_by_operation_id(self, operation_id: UUID) -> AppDeployment \| None:
		pass

		@abstractmethod
		async def save(self, app_deployment: AppDeployment) -> AppDeployment:
		pass

tests/unit/test_app_deployment_repo.py

0 → 100644

+104 −0

Original line number	Diff line number	Diff line
		from unittest.mock import AsyncMock
		from uuid import UUID, uuid4

		import pytest
		from sqlalchemy.ext.asyncio import AsyncSession

		from open_exposure_gateway.adapters.database.repos.deployments import (
		SqlAppDeploymentRepository,
		)
		from open_exposure_gateway.adapters.database.sql import AppDeploymentRow
		from open_exposure_gateway.domain.models import AppDeployment, AppDeploymentState


		def _row() -> AppDeploymentRow:
		return AppDeploymentRow(
		app_deployment_id=uuid4(),
		operation_id=uuid4(),
		app_registration_id=uuid4(),
		app_deployment_name="video_analytics_eu",
		edge_cloud_zones=[str(uuid4()), str(uuid4())],
		state=AppDeploymentState.INSTANTIATING,
		)


		def _domain() -> AppDeployment:
		return AppDeployment(
		app_deployment_id=uuid4(),
		operation_id=uuid4(),
		app_registration_id=uuid4(),
		app_deployment_name="video_analytics_eu",
		edge_cloud_zones=[uuid4(), uuid4()],
		state=AppDeploymentState.INSTANTIATING,
		)


		async def test_get_by_id_returns_mapped_app_deployment() -> None:
		row = _row()
		session = AsyncMock(spec=AsyncSession)
		session.scalar.return_value = row
		repo = SqlAppDeploymentRepository(session)

		result = await repo.get_by_id(row.app_deployment_id)

		assert result is not None
		assert result.app_deployment_id == row.app_deployment_id
		assert result.edge_cloud_zones == [UUID(z) for z in row.edge_cloud_zones]
		session.scalar.assert_awaited_once()


		async def test_get_by_id_returns_none_when_missing() -> None:
		session = AsyncMock(spec=AsyncSession)
		session.scalar.return_value = None
		repo = SqlAppDeploymentRepository(session)

		assert await repo.get_by_id(uuid4()) is None


		async def test_get_by_operation_id_returns_mapped_app_deployment() -> None:
		row = _row()
		session = AsyncMock(spec=AsyncSession)
		session.scalar.return_value = row
		repo = SqlAppDeploymentRepository(session)

		result = await repo.get_by_operation_id(row.operation_id)

		assert result is not None
		assert result.operation_id == row.operation_id


		async def test_get_by_operation_id_returns_none_when_missing() -> None:
		session = AsyncMock(spec=AsyncSession)
		session.scalar.return_value = None
		repo = SqlAppDeploymentRepository(session)

		assert await repo.get_by_operation_id(uuid4()) is None


		async def test_save_flushes_and_reloads_app_deployment() -> None:
		domain = _domain()
		expected = _domain()
		expected.app_deployment_id = domain.app_deployment_id
		session = AsyncMock(spec=AsyncSession)
		merged = _row()
		merged.app_deployment_id = domain.app_deployment_id
		session.merge.return_value = merged
		repo = SqlAppDeploymentRepository(session)
		repo.get_by_id = AsyncMock(return_value=expected) # type: ignore[method-assign]

		result = await repo.save(domain)

		assert result == expected
		session.merge.assert_awaited_once()
		session.flush.assert_awaited_once()
		repo.get_by_id.assert_awaited_once_with(domain.app_deployment_id)


		async def test_save_raises_when_reloaded_deployment_missing() -> None:
		session = AsyncMock(spec=AsyncSession)
		session.merge.return_value = _row()
		repo = SqlAppDeploymentRepository(session)
		repo.get_by_id = AsyncMock(return_value=None) # type: ignore[method-assign]

		with pytest.raises(RuntimeError):
		await repo.save(_domain())