Commit fd382ddf authored by George Papathanail's avatar George Papathanail Committed by Dimitrios Gogos
Browse files

feat: add SRM client read methods

parent 855bf088

src/open_exposure_gateway/adapters/http/srm_client.py

+31 −0
Original line number Diff line number Diff line
@@ -5,6 +5,9 @@ import httpx 
import structlog

from pydantic import ValidationError



from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (

    TrafficInfluence,

)

from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.core.exceptions import (

    DownstreamServiceException,
@@ -199,6 +202,34 @@ class SRMClient: 
        )

        return SRMNetworkCapability.model_validate(data)



    # NOTE: /internal/network-capabilities is not documented in

    # architecture/srm/interface-contract.md §E as a general capability read.

    # Confirm the real path with the SRM dev before relying on this in production.

    async def get_traffic_influence(

        self,

        traffic_influence_id: str,

        x_correlator: str | None = None,

    ) -> TrafficInfluence:

        headers = {"x-correlator": x_correlator} if x_correlator else None

        data = await self._request(

            "GET", f"/internal/network-capabilities/{traffic_influence_id}", headers=headers

        )

        return TrafficInfluence.model_validate(data)



    async def get_traffic_influences(

        self,

        app_id: UUID | None = None,

        x_correlator: str | None = None,

    ) -> list[TrafficInfluence]:

        params = {"capability_type": "traffic_influence"}

        if app_id is not None:

            params["app_id"] = str(app_id)

        headers = {"x-correlator": x_correlator} if x_correlator else None

        data = await self._request(

            "GET", "/internal/network-capabilities", params=params, headers=headers

        )

        return [TrafficInfluence.model_validate(t) for t in data]



    async def get_apps(self, x_correlator: str | None = None) -> list[SRMCatalogPayload]:

        headers = {"x-correlator": x_correlator} if x_correlator else None

        data = await self._request(

src/open_exposure_gateway/ports/srm_port.py

+14 −0
Original line number Diff line number Diff line 
from typing import Any, Protocol

from uuid import UUID



from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (

    TrafficInfluence,

)

from open_exposure_gateway.domain.edge_application_management import (

    SRMCatalogPayload,

    SRMCatalogServiceSpecificationCreated,
@@ -47,3 +50,14 @@ class SRMClientPort(Protocol): 
    async def retrieve_location(

        self, query: SRMLocationQuery, x_correlator: str | None

    ) -> SRMLocationResult: ...



    # NOTE: no such SRM endpoint is documented in architecture/srm/interface-contract.md

    # §E (only service-instances-by-id and operations-by-id exist there) -- this is an

    # open OEG<->SRM contract gap. Do not treat this as a confirmed wire shape.

    async def get_traffic_influence(

        self, traffic_influence_id: str, x_correlator: str | None

    ) -> TrafficInfluence: ...



    async def get_traffic_influences(

        self, app_id: UUID | None, x_correlator: str | None

    ) -> list[TrafficInfluence]: ...