hub-core/hub_core/runtime/ports.py

from __future__ import annotations

from uuid import UUID

from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
from jsonschema import ValidationError

from hub_core.runtime.models import (
    EventCommand,
    MessageCommand,
    PortAccepted,
    PortCollection,
    PortRecord,
    RegistryRegistration,
)
from hub_core.runtime.store import PortStore
from hub_core.runtime.validation import ContractValidator


def get_port_store(request: Request) -> PortStore:
    return request.app.state.port_store


def get_contract_validator(request: Request) -> ContractValidator:
    return request.app.state.contract_validator


def create_ports_router() -> APIRouter:
    router = APIRouter(prefix="/ports")

    @router.post(
        "/registry/registrations",
        response_model=PortAccepted,
        status_code=status.HTTP_202_ACCEPTED,
        tags=["registry"],
        openapi_extra={"x-port-id": "port.registry", "x-direction": "in"},
    )
    async def register_extension(
        body: RegistryRegistration,
        x_correlation_id: UUID = Header(alias="X-Correlation-ID"),
        store: PortStore = Depends(get_port_store),
        validator: ContractValidator = Depends(get_contract_validator),
    ) -> PortAccepted:
        try:
            validator.validate_registration(body)
        except (ValidationError, ValueError) as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc
        return await store.register_extension(body, x_correlation_id)

    @router.get(
        "/messaging/messages",
        response_model=PortCollection,
        tags=["messaging"],
        openapi_extra={"x-port-id": "port.messaging", "x-direction": "out"},
    )
    async def list_messages(
        address: str,
        conversation_id: UUID | None = None,
        store: PortStore = Depends(get_port_store),
    ) -> PortCollection:
        return await store.list_messages(address, conversation_id)

    @router.post(
        "/messaging/messages",
        response_model=PortAccepted,
        status_code=status.HTTP_202_ACCEPTED,
        tags=["messaging"],
        openapi_extra={"x-port-id": "port.messaging", "x-direction": "in"},
    )
    async def send_message(
        body: MessageCommand,
        store: PortStore = Depends(get_port_store),
    ) -> PortAccepted:
        return await store.send_message(body)

    @router.post(
        "/events/progress",
        response_model=PortAccepted,
        status_code=status.HTTP_202_ACCEPTED,
        tags=["events"],
        openapi_extra={"x-port-id": "port.events.progress", "x-direction": "in"},
    )
    async def append_progress(
        body: EventCommand,
        store: PortStore = Depends(get_port_store),
        validator: ContractValidator = Depends(get_contract_validator),
    ) -> PortAccepted:
        try:
            validator.validate_event_family(body.event_type, "progress")
        except ValueError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc
        return await store.append_progress(body)

    @router.post(
        "/events/interaction",
        response_model=PortAccepted,
        status_code=status.HTTP_202_ACCEPTED,
        tags=["events"],
        openapi_extra={"x-port-id": "port.events.interaction", "x-direction": "in"},
    )
    async def append_interaction(
        body: EventCommand,
        store: PortStore = Depends(get_port_store),
        validator: ContractValidator = Depends(get_contract_validator),
    ) -> PortAccepted:
        try:
            validator.validate_event_family(body.event_type, "interaction")
        except ValueError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc
        return await store.append_interaction(body)

    @router.get(
        "/projections/{projection_id}",
        response_model=PortRecord,
        tags=["projections"],
        openapi_extra={"x-port-id": "port.projection.query", "x-direction": "out"},
    )
    async def query_projection(
        projection_id: str,
        store: PortStore = Depends(get_port_store),
    ) -> PortRecord:
        projection = await store.query_projection(projection_id)
        if projection is None:
            raise HTTPException(status_code=404, detail=f"Projection '{projection_id}' not found")
        return projection

    return router