Commit b81c6852 authored by George Papathanail's avatar George Papathanail
Browse files

Merge branch 'feat/qod-v1' into 'develop' 

Feat/qod v1

See merge request !34
parents 32938d0b d8892b53

src/open_exposure_gateway/adapters/database/core.py

+13 −0
Changes for src/open_exposure_gateway/adapters/database/core.py: 13 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from typing import Literal



from sqlalchemy import text

from sqlalchemy.ext.asyncio import (

    AsyncEngine,

    AsyncSession,
@@ -23,6 +24,18 @@ async def build_engine_and_session_maker( 
    return engine, session_maker





# create_all never alters existing tables; columns added after first release go here.

_ADDED_COLUMNS = (

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS started_at TIMESTAMP WITH TIME ZONE",

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS ended_at TIMESTAMP WITH TIME ZONE",

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS status_info VARCHAR(32)",

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS device JSONB",

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS application_server JSONB",

)





async def schema_initialization(engine: AsyncEngine) -> None:

    async with engine.begin() as conn:

        await conn.run_sync(get_metadata().create_all)

        for statement in _ADDED_COLUMNS:

            await conn.execute(text(statement))

src/open_exposure_gateway/adapters/database/mappers.py

+12 −0
Changes for src/open_exposure_gateway/adapters/database/mappers.py: 12 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from typing import cast

from uuid import UUID



from open_exposure_gateway.adapters.database.sql import (
@@ -22,6 +23,7 @@ from open_exposure_gateway.domain.models import ( 
    QodSession,

    TrafficInfluence,

)

from open_exposure_gateway.domain.models.qod_sessions.models import QodStatusInfo





class AppRegistrationMapper:
@@ -157,6 +159,11 @@ class QodSessionMapper: 
            external_ref=row.external_ref,

            device_ports=row.device_ports,

            application_server_ports=row.application_server_ports,

            device=row.device,

            application_server=row.application_server,

            started_at=row.started_at,

            ended_at=row.ended_at,

            status_info=cast(QodStatusInfo | None, row.status_info),

            created_at=row.created_at,

            updated_at=row.updated_at,

        )
@@ -173,6 +180,11 @@ class QodSessionMapper: 
            external_ref=domain.external_ref,

            device_ports=domain.device_ports,

            application_server_ports=domain.application_server_ports,

            device=domain.device,

            application_server=domain.application_server,

            started_at=domain.started_at,

            ended_at=domain.ended_at,

            status_info=domain.status_info,

        )

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

+68 −3
Changes for src/open_exposure_gateway/adapters/database/repos/qod_sessions.py: 68 added lines, 3 removed lines.
Original line number Diff line number Diff line 
from datetime import datetime

from typing import Any

from uuid import UUID



from sqlalchemy import select

from sqlalchemy import and_, literal_column, or_, select, type_coerce

from sqlalchemy.dialects.postgresql import JSONB

from sqlalchemy.ext.asyncio import AsyncSession



from open_exposure_gateway.adapters.database.mappers import QodSessionMapper

from open_exposure_gateway.adapters.database.sql import QodSessionRow

from open_exposure_gateway.domain.models import QodSession

from open_exposure_gateway.adapters.database.sql import OperationRow, QodSessionRow

from open_exposure_gateway.domain.models import QodSession, QodSessionState

from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository



_ACTIVE_STATES = (QodSessionState.REQUESTED, QodSessionState.AVAILABLE)

_DEVICE_IDENTIFIERS = ("phoneNumber", "ipv4Address", "ipv6Address")





class SqlQodSessionRepository(QodSessionRepository):

    def __init__(self, session: AsyncSession) -> None:
@@ -23,6 +29,65 @@ class SqlQodSessionRepository(QodSessionRepository): 
        row = await self._session.scalar(stmt)

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



    async def find_active_by_device(

        self, device: dict[str, Any], app_provider_id: str

    ) -> QodSession | None:

        identifier_matches = [

            QodSessionRow.device[key] == type_coerce(device[key], JSONB)

            for key in _DEVICE_IDENTIFIERS

            if device.get(key) is not None

        ]

        if not identifier_matches:

            return None

        stmt = (

            select(QodSessionRow)

            .join(OperationRow, OperationRow.operation_id == QodSessionRow.operation_id)

            .where(

                OperationRow.app_provider_id == app_provider_id,

                or_(

                    QodSessionRow.state.in_(_ACTIVE_STATES),

                    and_(

                        QodSessionRow.state == QodSessionState.UNAVAILABLE,

                        QodSessionRow.status_info == "NETWORK_TERMINATED",

                    ),

                ),

                or_(*identifier_matches),

            )

            .limit(1)

        )

        row = await self._session.scalar(stmt)

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



    async def list_available_due(self, now: datetime) -> list[QodSession]:

        ends_at = QodSessionRow.started_at + QodSessionRow.duration_seconds * literal_column(

            "INTERVAL '1 second'"

        )

        stmt = (

            select(QodSessionRow)

            .where(

                QodSessionRow.state == QodSessionState.AVAILABLE,

                QodSessionRow.started_at.is_not(None),

                ends_at <= now,

            )

            .order_by(QodSessionRow.started_at)

        )

        rows = await self._session.scalars(stmt)

        return [QodSessionMapper.to_domain(row) for row in rows]



    async def list_network_terminated(self, ended_before: datetime) -> list[QodSession]:

        stmt = (

            select(QodSessionRow)

            .where(

                QodSessionRow.state == QodSessionState.UNAVAILABLE,

                QodSessionRow.status_info == "NETWORK_TERMINATED",

                QodSessionRow.ended_at.is_not(None),

                QodSessionRow.ended_at <= ended_before,

            )

            .order_by(QodSessionRow.ended_at)

        )

        rows = await self._session.scalars(stmt)

        return [QodSessionMapper.to_domain(row) for row in rows]



    async def save(self, qod_session: QodSession) -> QodSession:

        merged = await self._session.merge(QodSessionMapper.to_row(qod_session))

        await self._session.flush()

src/open_exposure_gateway/adapters/database/sql.py

+5 −0
Changes for src/open_exposure_gateway/adapters/database/sql.py: 5 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -244,6 +244,11 @@ class QodSessionRow(AuditedMixin, Base): 
    external_ref: Mapped[str | None] = mapped_column(String(255))

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

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

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

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

    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    status_info: Mapped[str | None] = mapped_column(String(32))





class TrafficInfluenceRow(AuditedMixin, Base):

src/open_exposure_gateway/adapters/http/qod_callback_client.py

+12 −3
Changes for src/open_exposure_gateway/adapters/http/qod_callback_client.py: 12 added lines, 3 removed lines.
Original line number Diff line number Diff line 
import httpx



from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.domain.errors import CallbackSinkGoneError

from open_exposure_gateway.domain.quality_on_demand import QosStatusChangedCloudEvent
@@ -10,11 +11,19 @@ class HttpQodCallbackClient: 
        self.timeout = settings.callback_settings.timeout



    async def deliver(

        self, sink: str, event: QosStatusChangedCloudEvent, bearer_token: str | None = None

        self,

        sink: str,

        event: QosStatusChangedCloudEvent,

        x_correlator: str,

        bearer_token: str | None = None,

    ) -> None:

        headers = {"Content-Type": "application/cloudevents+json"}

        headers = {"Content-Type": "application/cloudevents+json", "x-correlator": x_correlator}

        if bearer_token is not None:

            headers["Authorization"] = f"Bearer {bearer_token}"

        async with httpx.AsyncClient(timeout=self.timeout) as client:

            response = await client.post(sink, content=event.model_dump_json(), headers=headers)

            response = await client.post(

                sink, content=event.model_dump_json(exclude_none=True), headers=headers

            )

        if response.status_code == httpx.codes.GONE:

            raise CallbackSinkGoneError(f"sink answered {response.status_code}")

        response.raise_for_status()