Commit 3f9075ba authored by George Papathanail's avatar George Papathanail
Browse files

feat: Idempotency-key header and idempotent replay

parent 6a8b3f32

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

+5 −0
Original line number Diff line number Diff line
@@ -18,6 +18,11 @@ class SqlAppInstanceRepository(AppInstanceRepository): 
        row = await self._session.scalar(stmt)

        return AppInstanceMapper.to_domain(row) if row is not None else None



    async def get_by_operation_id(self, operation_id: UUID) -> AppInstance | None:

        stmt = select(AppInstanceRow).where(AppInstanceRow.operation_id == operation_id)

        row = await self._session.scalar(stmt)

        return AppInstanceMapper.to_domain(row) if row is not None else None



    async def save(self, app_instance: AppInstance) -> AppInstance:

        merged = await self._session.merge(AppInstanceMapper.to_row(app_instance))

        await self._session.flush()

src/open_exposure_gateway/api/camara/common.py

+5 −0
Original line number Diff line number Diff line
@@ -10,3 +10,8 @@ XCorrelatorHeader = Annotated[ 
    Optional[str],

    Header(alias="x-correlator", max_length=256, pattern=X_CORRELATOR_PATTERN_STR),

]



IdempotencyKeyHeader = Annotated[

    Optional[str],

    Header(alias="Idempotency-Key", max_length=128),

]

src/open_exposure_gateway/api/camara/edge_application_management/vwip/router.py

+3 −0
Original line number Diff line number Diff line
@@ -3,6 +3,7 @@ from uuid import UUID, uuid4 


from fastapi import APIRouter, Depends, Query, Request, Response, status



from open_exposure_gateway.api.camara.common import IdempotencyKeyHeader

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

    AppInstanceInfo,

    AppManifest,
@@ -170,12 +171,14 @@ async def create_app_instance( 
    caller: Caller,

    http_request: Request,

    response: Response,

    idempotency_key: IdempotencyKeyHeader = None,

) -> Any:

    instance = await service.create_app_instance(

        request=request,

        tenant_id=caller.tenant_id,

        app_provider_id=caller.app_provider_id,

        x_correlator=caller.x_correlator,

        idempotency_key=idempotency_key,

    )

    # Absolute URI per the spec ("Contains the URI of the newly created application.");

    # base_url reflects the scheme/host the caller actually used to reach OEG.

src/open_exposure_gateway/application/services/edge_application_management_service.py

+26 −0
Original line number Diff line number Diff line
@@ -206,11 +206,35 @@ class EdgeApplicationManagementService: 
        tenant_id: str,

        app_provider_id: str,

        x_correlator: Optional[str] = None,

        idempotency_key: Optional[str] = None,

    ) -> AppInstanceInfo:

        if self._app_registration_repo is None:

            raise RuntimeError("AppRegistrationRepository is not available")

        if self._operation_repo is None or self._app_instance_repo is None:

            raise RuntimeError("Operation/AppInstance repositories are not available")



        if idempotency_key is not None:

            existing_operation = await self._operation_repo.get_by_idempotency_key(

                tenant_id, idempotency_key

            )

            if existing_operation is not None:

                existing_instance = await self._app_instance_repo.get_by_operation_id(

                    existing_operation.operation_id

                )

                if existing_instance is None:

                    raise RuntimeError(

                        "operations row found for idempotency_key but its app_instances "

                        "row is missing"

                    )

                return AppInstanceInfo(

                    appInstanceId=existing_instance.app_instance_id,

                    name=request.name,

                    appId=request.appId,

                    appProvider=app_provider_id,

                    status=AppInstanceStatus(existing_instance.state),

                    edgeCloudZoneId=request.edgeCloudZoneId,

                )



        app_registration = await self._app_registration_repo.get_by_app_id(request.appId)

        if app_registration is None:

            raise BadRequestException(message=f"App {request.appId} is not registered")
@@ -224,6 +248,7 @@ class EdgeApplicationManagementService: 
            tenant_id=tenant_id,

            app_provider_id=app_provider_id,

            correlation_id=correlation_id,

            idempotency_key=idempotency_key,

        )



        await self._operation_repo.save(
@@ -235,6 +260,7 @@ class EdgeApplicationManagementService: 
                operation_type=OperationType.DEPLOY,

                status=OperationStatus.PENDING,

                subject=Subject.TASK_DEPLOY,

                idempotency_key=idempotency_key,

                app_registration_id=app_registration.app_registration_id,

                metadata={

                    "name": translation.name,

src/open_exposure_gateway/ports/database/instances.py

+4 −0
Original line number Diff line number Diff line
@@ -11,6 +11,10 @@ class AppInstanceRepository(ABC): 
    async def get_by_id(self, app_instance_id: UUID) -> AppInstance | None:

        pass



    @abstractmethod

    async def get_by_operation_id(self, operation_id: UUID) -> AppInstance | None:

        pass



    @abstractmethod

    async def save(self, app_instance: AppInstance) -> AppInstance:

        pass