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

from datetime import datetime
from typing import TYPE_CHECKING, Literal

from pydantic import BaseModel

if TYPE_CHECKING:
    from app.db.models import ApiKey, AppApiKey, AppDefinition, AppInvocationAudit, 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
    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):
    total_request_count: int
    total_error_count: int
    services: list[GatewayAuditServiceStats]


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


class ApiKeyListRequest(BaseModel):
    pass


class ApiKeyCreateResponse(BaseModel):
    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
    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):
    status: ApiKeyStatus


class ApiKeyStatusPostRequest(ApiKeyStatusUpdateRequest):
    api_key_id: str


# ── Application ──────────────────────────────────────────────────────────────

AppStatus = Literal["draft", "published", "disabled"]
AppTargetType = Literal["agent", "team"]


class AppCreateRequest(BaseModel):
    code: str
    name: str
    description: str | None = None
    target_type: AppTargetType
    target_id: str
    owner_user_id: str | None = None
    settings_json: str | None = None


class AppListRequest(BaseModel):
    pass


class AppDetailRequest(BaseModel):
    app_id: str


class AppUpdateRequest(BaseModel):
    app_id: str
    name: str | None = None
    description: str | None = None
    target_type: AppTargetType | None = None
    target_id: str | None = None
    settings_json: str | None = None


class AppStatusUpdateRequest(BaseModel):
    app_id: str
    status: AppStatus


class AppResponse(BaseModel):
    id: str
    code: str
    name: str
    description: str | None = None
    status: str
    target_type: str
    target_id: str
    owner_user_id: str | None = None
    settings_json: str | None = None
    created_time: datetime
    updated_time: datetime

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


# ── App API Key ──────────────────────────────────────────────────────────────

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


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


class AppApiKeyCreateResponse(BaseModel):
    id: str
    app_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 AppApiKeyListRequest(BaseModel):
    pass


class AppApiKeyResponse(BaseModel):
    id: str
    app_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: "AppApiKey") -> "AppApiKeyResponse":
        return cls.model_validate(entity, from_attributes=True)


class AppApiKeyStatusUpdateRequest(BaseModel):
    api_key_id: str
    status: AppApiKeyStatus


# ── App Invocation Audit ─────────────────────────────────────────────────────

class AppAuditListRequest(BaseModel):
    limit: int = 100


class AppInvocationAuditResponse(BaseModel):
    id: str
    app_id: str
    api_key_prefix: str | None = None
    request_id: str
    session_id: str | None = None
    run_request_id: str | None = None
    target_type: str
    target_id: str
    invoke_type: str
    status: str
    duration_ms: int
    error_code: str | None = None
    error_message: str | None = None
    client_metadata_json: str | None = None
    created_time: datetime

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


# ── OpenAPI External Invocation ──────────────────────────────────────────────

class OpenApiChatRequest(BaseModel):
    user_id: str | None = None
    session_id: str | None = None
    message: str
    inputs: dict[str, object] | None = None
    metadata: dict[str, object] | None = None


class OpenApiChatResponse(BaseModel):
    request_id: str
    app_code: str
    session_id: str
    run_request_id: str
    target_type: str
    target_id: str
    status: str
    output_text: str | None = None
    output_json: dict[str, object] | None = None
    error: str | None = None