Commit 52763755 authored by Paris Stentoumis's avatar Paris Stentoumis
Browse files

Merge branch 'refactor/fm-client' into 'refactor/tf-sdk-ports' 

SRM Refactor: Federation Manager Client ports and adapters

See merge request !28
parents 3e9f55cb 900cf5c4

.env.example

+3 −0
Changes for .env.example: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -10,3 +10,6 @@ NATS_SETTINGS__URL = "nats://localhost:4222" 
NATS_SETTINGS__CONNECT_TIMEOUT = 10

NATS_SETTINGS__MAX_RECONNECT_ATTEMPTS = 3

NATS_SETTINGS__DRAIN_TIMEOUT = 30



FEDERATION_MANAGER_SETTINGS__BASE_URL = "http://localhost:8082"

FEDERATION_MANAGER_SETTINGS__TIMEOUT = 0.5

src/srm/adapters/database/mappers.py

+2 −0
Changes for src/srm/adapters/database/mappers.py: 2 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -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,

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

+106 −1
Changes for src/srm/adapters/database/repos/runtime_inventory.py: 106 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -14,7 +14,12 @@ 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,

    ServiceInstanceRow,

    ServiceOrderRow,

)

from srm.domain.models.runtime_inventory import (

    CapabilityInstance,

    ServiceInstance,
@@ -25,6 +30,7 @@ from srm.domain.models.runtime_inventory.enums import ( 
    ServiceInstanceState,

    ServiceOrderState,

)

from srm.domain.models.topology.enums import CapabilityKind

from srm.domain.ports.database.runtime_inventory import (

    CapabilityInstanceRepository,

    ServiceInstanceRepository,
@@ -96,6 +102,21 @@ class SqlServiceOrderRepository(ServiceOrderRepository): 
            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:
@@ -112,6 +133,29 @@ 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,
@@ -201,3 +245,64 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository): 
        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]

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
@@ -270,6 +270,9 @@ class ServiceOrderRow(AuditedMixin, Base): 
    )

    state: Mapped[ServiceOrderState] = mapped_column(_enum_type(ServiceOrderState), nullable=False)

    payload_snapshot: Mapped[dict[str, object]] = mapped_column(JSONB, nullable=False, default=dict)

    federated_handoff: Mapped[dict[str, object]] = mapped_column(

        JSONB, nullable=False, default=dict

    )

    failure_detail: Mapped[dict[str, object] | None] = mapped_column(JSONB)

    app_provider_id: Mapped[str] = mapped_column(String(255), nullable=False)

    federation_partner_ref: Mapped[str | None] = mapped_column(String(255))

src/srm/adapters/errors.py

+12 −1
Changes for src/srm/adapters/errors.py: 12 added lines, 1 removed line.
Original line number Diff line number Diff line 
from typing import ClassVar

from typing import Any, ClassVar





class DuplicateEntryError(Exception):
@@ -15,3 +15,14 @@ class DuplicateServiceSpecificationError(DuplicateEntryError): 
class ServiceSpecificationInUseError(Exception):

    def __init__(self) -> None:

        super().__init__("Service specification has live service instances.")





class NotFoundException(Exception):

    def __init__(self) -> None:

        super().__init__("Resource Not Found")





class DownstreamServiceException(Exception):

    def __init__(self, message: str = "Service Unavailable", details: Any | None = None):

        super().__init__(message)

        self._details = details