Commit 4b7afb76 authored by George Papathanail's avatar George Papathanail
Browse files

feat: start event.srm.operation.status consumer, wire callback client

parent 458ae0eb

src/open_exposure_gateway/domain/srm_events.py

+4 −2
Original line number Diff line number Diff line
@@ -45,8 +45,10 @@ class SRMOperationCompleted(BaseModel): 


    @model_validator(mode="after")

    def _require_error_when_failed(self) -> SRMOperationCompleted:

        if self.status == "failed" and self.error is None:

            raise ValueError("error is required when status is failed")

        if self.status == "failed" and not self.instances and self.error is None:

            raise ValueError(

                "error is required when status is failed and no instances were produced"

            )

        return self



    @model_validator(mode="after")

src/open_exposure_gateway/main.py

+56 −9
Original line number Diff line number Diff line
@@ -28,6 +28,7 @@ from open_exposure_gateway.adapters.databus.nats_adapter import ( 
    NatsOperationConsumer,

)

from open_exposure_gateway.adapters.http.callback_client import HttpCallbackClient

from open_exposure_gateway.adapters.http.qod_callback_client import HttpQodCallbackClient

from open_exposure_gateway.adapters.http.srm_client import SRMClient

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

    router as edge_application_management_router,
@@ -49,8 +50,10 @@ from open_exposure_gateway.application.services.quality_on_demand_service import 
from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.core.logging import configure_logging

from open_exposure_gateway.domain.edge_application_management import Subject

from open_exposure_gateway.domain.srm_events import SRMOperationCompleted

from open_exposure_gateway.domain.quality_on_demand import Subject as QodSubject

from open_exposure_gateway.domain.srm_events import SRMOperationCompleted, SRMOperationStatus

from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort

from open_exposure_gateway.ports.qod_callback_port import QodCallbackDeliveryPort

from open_exposure_gateway.ports.srm_port import SRMClientPort
@@ -80,18 +83,30 @@ def _build_operation_completed_handler( 
    return handle





def _build_qod_service(

    session: AsyncSession,

    srm_client: SRMClientPort,

    qod_callback_client: QodCallbackDeliveryPort,

) -> QualityOnDemandService:

    return QualityOnDemandService(

        srm_client=srm_client,

        operation_repo=SqlOperationRepository(session),

        qod_session_repo=SqlQodSessionRepository(session),

        callback_registration_repo=SqlCallbackRegistrationRepository(session),

        callback_delivery_repo=SqlCallbackDeliveryRepository(session),

        callback_delivery_port=qod_callback_client,

    )





def _build_qod_operation_completed_handler(

    session_maker: async_sessionmaker[AsyncSession],

    srm_client: SRMClientPort,

    qod_callback_client: QodCallbackDeliveryPort,

) -> Callable[[SRMOperationCompleted], Awaitable[None]]:

    async def handle(event: SRMOperationCompleted) -> None:

        async with session_maker() as session:

            try:

                service = QualityOnDemandService(

                    srm_client=srm_client,

                    operation_repo=SqlOperationRepository(session),

                    qod_session_repo=SqlQodSessionRepository(session),

                )

                service = _build_qod_service(session, srm_client, qod_callback_client)

                await service.handle_completed(event)

                await session.commit()

            except Exception:
@@ -101,6 +116,24 @@ def _build_qod_operation_completed_handler( 
    return handle





def _build_qod_operation_status_handler(

    session_maker: async_sessionmaker[AsyncSession],

    srm_client: SRMClientPort,

    qod_callback_client: QodCallbackDeliveryPort,

) -> Callable[[SRMOperationStatus], Awaitable[None]]:

    async def handle(event: SRMOperationStatus) -> None:

        async with session_maker() as session:

            try:

                service = _build_qod_service(session, srm_client, qod_callback_client)

                await service.handle_status_changed(event)

                await session.commit()

            except Exception:

                await session.rollback()

                raise



    return handle





openapi_tags = [

    {

        "name": "Application",
@@ -154,6 +187,8 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 
        logger.error("Failed to initialize SRM client", error=str(e))

        raise



    qod_callback_client = HttpQodCallbackClient()



    try:

        publisher = NatsMessagePublisher(settings.nats_settings)

        await publisher.connect()
@@ -169,18 +204,30 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 
    )

    await eam_operation_consumer.start()



    qod_operation_consumer = NatsOperationConsumer(

    operation_completed_consumer = NatsOperationConsumer(

        client=publisher.client,

        subject=Subject.OPERATION_COMPLETED,

        handler=_build_qod_operation_completed_handler(session_maker, srm_client),

        handler=_build_qod_operation_completed_handler(

            session_maker, srm_client, qod_callback_client

        ),

    )

    await qod_operation_consumer.start()

    await operation_completed_consumer.start()

    logger.info("NATS consumer started", subject=Subject.OPERATION_COMPLETED)



    operation_status_consumer = NatsOperationConsumer(

        client=publisher.client,

        subject=QodSubject.OPERATION_STATUS,

        event_model=SRMOperationStatus,

        handler=_build_qod_operation_status_handler(session_maker, srm_client, qod_callback_client),

    )

    await operation_status_consumer.start()

    logger.info("NATS consumer started", subject=QodSubject.OPERATION_STATUS)



    app.state.srm_client = srm_client

    app.state.publisher = publisher

    app.state.db_engine = db_engine

    app.state.session_maker = session_maker

    app.state.qod_callback_client = qod_callback_client



    yield