quality-inspection-platform/app/schemas.py

from typing import Any, Literal

from pydantic import BaseModel, Field


class ApiBase(BaseModel):
    name: str
    folder_path: str = ""
    method: str = "GET"
    url: str
    headers: dict[str, Any] = Field(default_factory=dict)
    body: dict[str, Any] = Field(default_factory=dict)
    query: dict[str, Any] = Field(default_factory=dict)
    path_params: dict[str, Any] = Field(default_factory=dict)
    timeout_seconds: float = 10.0


class ApiCreate(ApiBase):
    pass


class ApiUpdate(ApiBase):
    pass


class ApiOut(ApiBase):
    id: int
    creator_id: int = 0
    creator_name: str = ""

    class Config:
        from_attributes = True


class WorkflowPayload(BaseModel):
    nodes: list[dict[str, Any]] = Field(default_factory=list)
    edges: list[dict[str, Any]] = Field(default_factory=list)
    variables: dict[str, Any] = Field(default_factory=dict)


class WorkflowBase(BaseModel):
    name: str
    folder_path: str = ""
    definition: WorkflowPayload = Field(default_factory=WorkflowPayload)
    default_headers: dict[str, Any] = Field(default_factory=dict)


class WorkflowCreate(WorkflowBase):
    pass


class WorkflowUpdate(WorkflowBase):
    pass


class WorkflowOut(WorkflowBase):
    id: int
    creator_id: int = 0
    creator_name: str = ""
    last_run: dict[str, Any] | None = None

    class Config:
        from_attributes = True


class ReplayRunRequest(BaseModel):
    base_url: str = ""
    fail_fast: bool = True


class MockDataBase(BaseModel):
    name: str
    data: dict[str, Any] = Field(default_factory=dict)


class MockDataCreate(MockDataBase):
    pass


class MockDataOut(MockDataBase):
    id: int
    creator_id: int = 0
    creator_name: str = ""

    class Config:
        from_attributes = True


class McpToolBase(BaseModel):
    name: str
    enabled: bool = True
    config: dict[str, Any] = Field(default_factory=dict)


class McpToolCreate(McpToolBase):
    pass


class McpToolOut(McpToolBase):
    id: int
    creator_id: int = 0
    creator_name: str = ""

    class Config:
        from_attributes = True


class SshCommandPreset(BaseModel):
    name: str
    command: str
    description: str = ""


class SshProfileBase(BaseModel):
    name: str
    host: str
    port: int = 22
    username: str
    auth_type: Literal["key", "password"] = "key"
    key_path: str = ""
    presets: list[SshCommandPreset] = Field(default_factory=list)


class SshProfileCreate(SshProfileBase):
    pass


class SshProfileUpdate(SshProfileBase):
    pass


class SshProfileOut(SshProfileBase):
    id: int
    creator_id: int = 0
    creator_name: str = ""

    class Config:
        from_attributes = True


class SshExecuteRequest(BaseModel):
    command: str = ""
    preset_name: str = ""
    password: str = ""
    timeout_seconds: float = 20.0


class SshExecuteResponse(BaseModel):
    ok: bool
    command: str
    stdout: str = ""
    stderr: str = ""
    exit_status: int = 0


class SshShortcutBase(BaseModel):
    name: str
    command: str
    description: str = ""


class SshShortcutCreate(SshShortcutBase):
    pass


class SshShortcutUpdate(SshShortcutBase):
    pass


class SshShortcutOut(SshShortcutBase):
    id: int
    creator_id: int = 0
    creator_name: str = ""

    class Config:
        from_attributes = True


class SshScriptBase(BaseModel):
    profile_id: int
    name: str
    description: str = ""
    source_type: Literal["inline", "file"] = "inline"
    content: str = ""


class SshScriptCreate(SshScriptBase):
    pass


class SshScriptUpdate(BaseModel):
    name: str
    description: str = ""
    content: str = ""


class SshScriptOut(SshScriptBase):
    id: int
    storage_path: str = ""
    creator_id: int = 0
    creator_name: str = ""

    class Config:
        from_attributes = True


class SshTreeScriptNode(BaseModel):
    id: int
    name: str
    description: str = ""
    source_type: Literal["inline", "file"] = "inline"
    storage_path: str = ""


class SshTreeProfileNode(BaseModel):
    id: int
    name: str
    host: str
    port: int = 22
    username: str
    auth_type: Literal["key", "password"] = "key"
    key_path: str = ""
    scripts: list[SshTreeScriptNode] = Field(default_factory=list)


class LoginRequest(BaseModel):
    username: str
    password: str


class CurrentUserOut(BaseModel):
    id: int
    username: str
    display_name: str
    role: Literal["superadmin", "user"]


class LoginResponse(BaseModel):
    token: str
    user: CurrentUserOut


class UserCreate(BaseModel):
    username: str
    password: str
    display_name: str = ""
    role: Literal["superadmin", "user"] = "user"


class UserUpdate(BaseModel):
    display_name: str = ""
    role: Literal["superadmin", "user"] = "user"
    is_active: bool = True
    password: str = ""


class UserOut(BaseModel):
    id: int
    username: str
    display_name: str
    role: Literal["superadmin", "user"]
    is_active: bool


class UserApiKeyCreate(BaseModel):
    expires_at: str | None = None


class UserApiKeyOut(BaseModel):
    id: int
    key_prefix: str
    key_hint: str
    masked_key: str = ""
    expires_at: str | None = None
    is_active: bool = True
    created_at: str = ""

    class Config:
        from_attributes = True


class UserApiKeyCreateResponse(UserApiKeyOut):
    api_key: str


class RunWorkflowRequest(BaseModel):
    workflow_id: int
    base_url: str = ""
    fail_fast: bool = True


class NodeExecutionResult(BaseModel):
    node_id: str
    node_type: str
    status: Literal["success", "failed", "skipped"]
    output: dict[str, Any] = Field(default_factory=dict)
    error: str | None = None


class RunWorkflowResponse(BaseModel):
    status: Literal["success", "failed"]
    variables: dict[str, Any]
    results: list[NodeExecutionResult]
    run_id: int | None = None
    workflow_id: int | None = None
    batch_id: int | None = None


class WorkflowBatchCreate(BaseModel):
    name: str = ""
    base_url: str = ""
    fail_fast: bool = True
    workflow_ids: list[int] = Field(default_factory=list)


class WorkflowBatchUpdate(BaseModel):
    name: str | None = None
    base_url: str | None = None
    fail_fast: bool | None = None
    workflow_ids: list[int] | None = None


class RunWorkflowBatchRequest(BaseModel):
    """快捷：创建批次并立即执行（兼容旧调用）。"""
    workflow_ids: list[int] = Field(default_factory=list)
    name: str = ""
    base_url: str = ""
    fail_fast: bool = True


class WorkflowBatchRunItem(BaseModel):
    workflow_id: int
    workflow_name: str = ""
    run_id: int | None = None
    status: str = "pending"
    error: str | None = None


class RunWorkflowBatchResponse(BaseModel):
    batch_id: int
    status: Literal["draft", "running", "success", "failed", "partial"]
    items: list[WorkflowBatchRunItem]


class WorkflowBatchOut(BaseModel):
    id: int
    name: str = ""
    status: Literal["draft", "running", "success", "failed", "partial"]
    triggered_by: str = ""
    creator_id: int = 0
    creator_name: str = ""
    base_url: str = ""
    fail_fast: bool = True
    workflow_ids: list[int] = Field(default_factory=list)
    summary: dict[str, Any] = Field(default_factory=dict)
    created_at: str | None = None
    finished_at: str | None = None
    runs: list[dict[str, Any]] = Field(default_factory=list)


class LokiLogLinkOut(BaseModel):
    enabled: bool = False
    explore_url: str = ""
    logql: str = ""
    time_from: str = ""
    time_to: str = ""
    method: str = ""
    url: str = ""
    hint: str = ""


class RunNodeRequest(BaseModel):
    workflow_id: int
    node_id: str
    base_url: str = ""


class RunNodeResponse(BaseModel):
    status: Literal["success", "failed"]
    variables: dict[str, Any]
    result: NodeExecutionResult


class FolderMoveRequest(BaseModel):
    target: Literal["apis", "workflows"]
    from_path: str
    to_path: str


class McpInvokeRequest(BaseModel):
    tool: str
    arguments: dict[str, Any] = Field(default_factory=dict)
    api_key: str | None = Field(
        default=None,
        description="Optional sto- API key when Authorization header is unavailable",
    )


class McpInvokeBatchRequest(BaseModel):
    calls: list[McpInvokeRequest] = Field(default_factory=list)
    stop_on_error: bool = True


class McpInvokeResponse(BaseModel):
    ok: bool
    tool: str
    data: dict[str, Any] = Field(default_factory=dict)
    error: str | None = None