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