agent-platform/services/api-gateway/app/schemas/gateway.py

from pydantic import BaseModel
from datetime import datetime
from typing import Literal

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from app.db.models import ApiKey, GatewayRequestAudit


class DownstreamServiceHealth(BaseModel):
    service: str
    status: str
    url: str
    status_code: int | None = None
    error_message: str | None = None


class GatewayServicesHealthResponse(BaseModel):
    service: str = "api-gateway"
    status: str
    downstream_services: list[DownstreamServiceHealth]


class GatewayRequestAuditResponse(BaseModel):
    id: str
    tenant_id: str
    request_id: str
    method: str
    path: str
    query_string: str | None = None
    target_service: str | None = None
    target_url: str | None = None
    status_code: int | None = None
    duration_ms: int
    client_host: str | None = None
    user_agent: str | None = None
    error_message: str | None = None
    created_time: datetime

    @classmethod
    def from_entity(cls, entity: "GatewayRequestAudit") -> "GatewayRequestAuditResponse":
        return cls.model_validate(entity, from_attributes=True)


class GatewayAuditServiceStats(BaseModel):
    target_service: str
    request_count: int
    error_count: int
    average_duration_ms: float


class GatewayAuditStatsResponse(BaseModel):
    tenant_id: str
    total_request_count: int
    total_error_count: int
    services: list[GatewayAuditServiceStats]


class ApiKeyCreateRequest(BaseModel):
    tenant_id: str
    name: str
    scopes: str | None = None
    expires_time: datetime | None = None


class ApiKeyCreateResponse(BaseModel):
    id: str
    tenant_id: str
    name: str
    key_prefix: str
    api_key: str
    status: str
    scopes: str | None = None
    expires_time: datetime | None = None
    created_time: datetime


class ApiKeyResponse(BaseModel):
    id: str
    tenant_id: str
    name: str
    key_prefix: str
    status: str
    scopes: str | None = None
    expires_time: datetime | None = None
    last_used_time: datetime | None = None
    created_time: datetime

    @classmethod
    def from_entity(cls, entity: "ApiKey") -> "ApiKeyResponse":
        return cls.model_validate(entity, from_attributes=True)


ApiKeyStatus = Literal["active", "disabled", "revoked"]


class ApiKeyStatusUpdateRequest(BaseModel):
    tenant_id: str
    status: ApiKeyStatus