ComfyUI-Manager/comfyui_manager/data_models/generated_models.py

# generated by datamodel-codegen:
#   filename:  openapi.yaml
#   timestamp: 2025-06-27T04:01:45+00:00

from __future__ import annotations

from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional, Union

from pydantic import BaseModel, Field, RootModel


class OperationType(str, Enum):
    install = "install"
    uninstall = "uninstall"
    update = "update"
    update_comfyui = "update-comfyui"
    fix = "fix"
    disable = "disable"
    enable = "enable"
    install_model = "install-model"


class OperationResult(str, Enum):
    success = "success"
    failed = "failed"
    skipped = "skipped"
    error = "error"
    skip = "skip"


class TaskExecutionStatus(BaseModel):
    status_str: OperationResult
    completed: bool = Field(..., description="Whether the task completed")
    messages: List[str] = Field(..., description="Additional status messages")


class ManagerMessageName(str, Enum):
    cm_task_completed = "cm-task-completed"
    cm_task_started = "cm-task-started"
    cm_queue_status = "cm-queue-status"


class ManagerPackInfo(BaseModel):
    id: str = Field(
        ...,
        description="Either github-author/github-repo or name of pack from the registry",
    )
    version: str = Field(..., description="Semantic version or Git commit hash")
    ui_id: Optional[str] = Field(None, description="Task ID - generated internally")


class ManagerPackInstalled(BaseModel):
    ver: str = Field(
        ...,
        description="The version of the pack that is installed (Git commit hash or semantic version)",
    )
    cnr_id: Optional[str] = Field(
        None, description="The name of the pack if installed from the registry"
    )
    aux_id: Optional[str] = Field(
        None,
        description="The name of the pack if installed from github (author/repo-name format)",
    )
    enabled: bool = Field(..., description="Whether the pack is enabled")


class SelectedVersion(str, Enum):
    latest = "latest"
    nightly = "nightly"


class ManagerChannel(str, Enum):
    default = "default"
    recent = "recent"
    legacy = "legacy"
    forked = "forked"
    dev = "dev"
    tutorial = "tutorial"


class ManagerDatabaseSource(str, Enum):
    remote = "remote"
    local = "local"
    cache = "cache"


class ManagerPackState(str, Enum):
    installed = "installed"
    disabled = "disabled"
    not_installed = "not_installed"
    import_failed = "import_failed"
    needs_update = "needs_update"


class ManagerPackInstallType(str, Enum):
    git_clone = "git-clone"
    copy = "copy"
    cnr = "cnr"


class SecurityLevel(str, Enum):
    strong = "strong"
    normal = "normal"
    normal_ = "normal-"
    weak = "weak"


class RiskLevel(str, Enum):
    block = "block"
    high_ = "high+"
    high = "high"
    middle_ = "middle+"
    middle = "middle"


class UpdateState(Enum):
    false = "false"
    true = "true"


class ManagerPack(ManagerPackInfo):
    author: Optional[str] = Field(
        None, description="Pack author name or 'Unclaimed' if added via GitHub crawl"
    )
    files: Optional[List[str]] = Field(
        None,
        description="Repository URLs for installation (typically contains one GitHub URL)",
    )
    reference: Optional[str] = Field(
        None, description="The type of installation reference"
    )
    title: Optional[str] = Field(None, description="The display name of the pack")
    cnr_latest: Optional[SelectedVersion] = None
    repository: Optional[str] = Field(None, description="GitHub repository URL")
    state: Optional[ManagerPackState] = None
    update_state: Optional[UpdateState] = Field(
        None, alias="update-state", description="Update availability status"
    )
    stars: Optional[int] = Field(None, description="GitHub stars count")
    last_update: Optional[datetime] = Field(None, description="Last update timestamp")
    health: Optional[str] = Field(None, description="Health status of the pack")
    description: Optional[str] = Field(None, description="Pack description")
    trust: Optional[bool] = Field(None, description="Whether the pack is trusted")
    install_type: Optional[ManagerPackInstallType] = None


class InstallPackParams(ManagerPackInfo):
    selected_version: Union[str, SelectedVersion] = Field(
        ..., description="Semantic version, Git commit hash, latest, or nightly"
    )
    repository: Optional[str] = Field(
        None,
        description="GitHub repository URL (required if selected_version is nightly)",
    )
    pip: Optional[List[str]] = Field(None, description="PyPi dependency names")
    mode: ManagerDatabaseSource
    channel: ManagerChannel
    skip_post_install: Optional[bool] = Field(
        None, description="Whether to skip post-installation steps"
    )


class UpdateAllPacksParams(BaseModel):
    mode: Optional[ManagerDatabaseSource] = None
    ui_id: Optional[str] = Field(None, description="Task ID - generated internally")


class UpdatePackParams(BaseModel):
    node_name: str = Field(..., description="Name of the node package to update")
    node_ver: Optional[str] = Field(
        None, description="Current version of the node package"
    )


class UpdateComfyUIParams(BaseModel):
    is_stable: Optional[bool] = Field(
        True,
        description="Whether to update to stable version (true) or nightly (false)",
    )
    target_version: Optional[str] = Field(
        None,
        description="Specific version to switch to (for version switching operations)",
    )


class FixPackParams(BaseModel):
    node_name: str = Field(..., description="Name of the node package to fix")
    node_ver: str = Field(..., description="Version of the node package")


class UninstallPackParams(BaseModel):
    node_name: str = Field(..., description="Name of the node package to uninstall")
    is_unknown: Optional[bool] = Field(
        False, description="Whether this is an unknown/unregistered package"
    )


class DisablePackParams(BaseModel):
    node_name: str = Field(..., description="Name of the node package to disable")
    is_unknown: Optional[bool] = Field(
        False, description="Whether this is an unknown/unregistered package"
    )


class EnablePackParams(BaseModel):
    cnr_id: str = Field(
        ..., description="ComfyUI Node Registry ID of the package to enable"
    )


class UpdateAllQueryParams(BaseModel):
    client_id: str = Field(
        ..., description="Client identifier that initiated the request"
    )
    ui_id: str = Field(..., description="Base UI identifier for task tracking")
    mode: Optional[ManagerDatabaseSource] = None


class UpdateComfyUIQueryParams(BaseModel):
    client_id: str = Field(
        ..., description="Client identifier that initiated the request"
    )
    ui_id: str = Field(..., description="UI identifier for task tracking")
    stable: Optional[bool] = Field(
        True,
        description="Whether to update to stable version (true) or nightly (false)",
    )


class ComfyUISwitchVersionQueryParams(BaseModel):
    ver: str = Field(..., description="Version to switch to")
    client_id: str = Field(
        ..., description="Client identifier that initiated the request"
    )
    ui_id: str = Field(..., description="UI identifier for task tracking")


class QueueStatus(BaseModel):
    total_count: int = Field(
        ..., description="Total number of tasks (pending + running)"
    )
    done_count: int = Field(..., description="Number of completed tasks")
    in_progress_count: int = Field(..., description="Number of tasks currently running")
    pending_count: Optional[int] = Field(
        None, description="Number of tasks waiting to be executed"
    )
    is_processing: bool = Field(..., description="Whether the task worker is active")
    client_id: Optional[str] = Field(
        None, description="Client ID (when filtered by client)"
    )


class ManagerMappings1(BaseModel):
    title_aux: Optional[str] = Field(None, description="The display name of the pack")


class ManagerMappings(
    RootModel[Optional[Dict[str, List[Union[List[str], ManagerMappings1]]]]]
):
    root: Optional[Dict[str, List[Union[List[str], ManagerMappings1]]]] = Field(
        None, description="Tuple of [node_names, metadata]"
    )


class ModelMetadata(BaseModel):
    name: str = Field(..., description="Name of the model")
    type: str = Field(..., description="Type of model")
    base: Optional[str] = Field(None, description="Base model type")
    save_path: Optional[str] = Field(None, description="Path for saving the model")
    url: str = Field(..., description="Download URL")
    filename: str = Field(..., description="Target filename")
    ui_id: Optional[str] = Field(None, description="ID for UI reference")


class InstallType(str, Enum):
    git = "git"
    copy = "copy"
    pip = "pip"


class NodePackageMetadata(BaseModel):
    title: Optional[str] = Field(None, description="Display name of the node package")
    name: Optional[str] = Field(None, description="Repository/package name")
    files: Optional[List[str]] = Field(None, description="Source URLs for the package")
    description: Optional[str] = Field(
        None, description="Description of the node package functionality"
    )
    install_type: Optional[InstallType] = Field(None, description="Installation method")
    version: Optional[str] = Field(None, description="Version identifier")
    id: Optional[str] = Field(
        None, description="Unique identifier for the node package"
    )
    ui_id: Optional[str] = Field(None, description="ID for UI reference")
    channel: Optional[str] = Field(None, description="Source channel")
    mode: Optional[str] = Field(None, description="Source mode")


class SnapshotItem(RootModel[str]):
    root: str = Field(..., description="Name of the snapshot")


class Error(BaseModel):
    error: str = Field(..., description="Error message")


class InstalledPacksResponse(RootModel[Optional[Dict[str, ManagerPackInstalled]]]):
    root: Optional[Dict[str, ManagerPackInstalled]] = None


class HistoryListResponse(BaseModel):
    ids: Optional[List[str]] = Field(
        None, description="List of available batch history IDs"
    )


class InstalledNodeInfo(BaseModel):
    name: str = Field(..., description="Node package name")
    version: str = Field(..., description="Installed version")
    repository_url: Optional[str] = Field(None, description="Git repository URL")
    install_method: str = Field(
        ..., description="Installation method (cnr, git, pip, etc.)"
    )
    enabled: Optional[bool] = Field(
        True, description="Whether the node is currently enabled"
    )
    install_date: Optional[datetime] = Field(
        None, description="ISO timestamp of installation"
    )


class InstalledModelInfo(BaseModel):
    name: str = Field(..., description="Model filename")
    path: str = Field(..., description="Full path to model file")
    type: str = Field(..., description="Model type (checkpoint, lora, vae, etc.)")
    size_bytes: Optional[int] = Field(None, description="File size in bytes", ge=0)
    hash: Optional[str] = Field(None, description="Model file hash for verification")
    install_date: Optional[datetime] = Field(
        None, description="ISO timestamp when added"
    )


class ComfyUIVersionInfo(BaseModel):
    version: str = Field(..., description="ComfyUI version string")
    commit_hash: Optional[str] = Field(None, description="Git commit hash")
    branch: Optional[str] = Field(None, description="Git branch name")
    is_stable: Optional[bool] = Field(
        False, description="Whether this is a stable release"
    )
    last_updated: Optional[datetime] = Field(
        None, description="ISO timestamp of last update"
    )


class BatchOperation(BaseModel):
    operation_id: str = Field(..., description="Unique operation identifier")
    operation_type: OperationType
    target: str = Field(
        ..., description="Target of the operation (node name, model name, etc.)"
    )
    target_version: Optional[str] = Field(
        None, description="Target version for the operation"
    )
    result: OperationResult
    error_message: Optional[str] = Field(
        None, description="Error message if operation failed"
    )
    start_time: datetime = Field(
        ..., description="ISO timestamp when operation started"
    )
    end_time: Optional[datetime] = Field(
        None, description="ISO timestamp when operation completed"
    )
    client_id: Optional[str] = Field(
        None, description="Client that initiated the operation"
    )


class ComfyUISystemState(BaseModel):
    snapshot_time: datetime = Field(
        ..., description="ISO timestamp when snapshot was taken"
    )
    comfyui_version: ComfyUIVersionInfo
    frontend_version: Optional[str] = Field(
        None, description="ComfyUI frontend version if available"
    )
    python_version: str = Field(..., description="Python interpreter version")
    platform_info: str = Field(
        ..., description="Operating system and platform information"
    )
    installed_nodes: Optional[Dict[str, InstalledNodeInfo]] = Field(
        None, description="Map of installed node packages by name"
    )
    installed_models: Optional[Dict[str, InstalledModelInfo]] = Field(
        None, description="Map of installed models by name"
    )
    manager_config: Optional[Dict[str, Any]] = Field(
        None, description="ComfyUI Manager configuration settings"
    )
    comfyui_root_path: Optional[str] = Field(
        None, description="ComfyUI root installation directory"
    )
    model_paths: Optional[Dict[str, List[str]]] = Field(
        None, description="Map of model types to their configured paths"
    )
    manager_version: Optional[str] = Field(None, description="ComfyUI Manager version")
    security_level: Optional[SecurityLevel] = None
    network_mode: Optional[str] = Field(
        None, description="Network mode (online, offline, private)"
    )
    cli_args: Optional[Dict[str, Any]] = Field(
        None, description="Selected ComfyUI CLI arguments"
    )
    custom_nodes_count: Optional[int] = Field(
        None, description="Total number of custom node packages", ge=0
    )
    failed_imports: Optional[List[str]] = Field(
        None, description="List of custom nodes that failed to import"
    )
    pip_packages: Optional[Dict[str, str]] = Field(
        None, description="Map of installed pip packages to their versions"
    )
    embedded_python: Optional[bool] = Field(
        None,
        description="Whether ComfyUI is running from an embedded Python distribution",
    )


class BatchExecutionRecord(BaseModel):
    batch_id: str = Field(..., description="Unique batch identifier")
    start_time: datetime = Field(..., description="ISO timestamp when batch started")
    end_time: Optional[datetime] = Field(
        None, description="ISO timestamp when batch completed"
    )
    state_before: ComfyUISystemState
    state_after: Optional[ComfyUISystemState] = Field(
        None, description="System state after batch execution"
    )
    operations: Optional[List[BatchOperation]] = Field(
        None, description="List of operations performed in this batch"
    )
    total_operations: Optional[int] = Field(
        0, description="Total number of operations in batch", ge=0
    )
    successful_operations: Optional[int] = Field(
        0, description="Number of successful operations", ge=0
    )
    failed_operations: Optional[int] = Field(
        0, description="Number of failed operations", ge=0
    )
    skipped_operations: Optional[int] = Field(
        0, description="Number of skipped operations", ge=0
    )


class QueueTaskItem(BaseModel):
    ui_id: str = Field(..., description="Unique identifier for the task")
    client_id: str = Field(..., description="Client identifier that initiated the task")
    kind: OperationType
    params: Union[
        InstallPackParams,
        UpdatePackParams,
        UpdateAllPacksParams,
        UpdateComfyUIParams,
        FixPackParams,
        UninstallPackParams,
        DisablePackParams,
        EnablePackParams,
        ModelMetadata,
    ]


class TaskHistoryItem(BaseModel):
    ui_id: str = Field(..., description="Unique identifier for the task")
    client_id: str = Field(..., description="Client identifier that initiated the task")
    kind: str = Field(..., description="Type of task that was performed")
    timestamp: datetime = Field(..., description="ISO timestamp when task completed")
    result: str = Field(..., description="Task result message or details")
    status: Optional[TaskExecutionStatus] = None
    batch_id: Optional[str] = Field(
        None, description="ID of the batch this task belongs to"
    )
    end_time: Optional[datetime] = Field(
        None, description="ISO timestamp when task execution ended"
    )


class TaskStateMessage(BaseModel):
    history: Dict[str, TaskHistoryItem] = Field(
        ..., description="Map of task IDs to their history items"
    )
    running_queue: List[QueueTaskItem] = Field(
        ..., description="Currently executing tasks"
    )
    pending_queue: List[QueueTaskItem] = Field(
        ..., description="Tasks waiting to be executed"
    )
    installed_packs: Dict[str, ManagerPackInstalled] = Field(
        ..., description="Map of currently installed node packages by name"
    )


class MessageTaskDone(BaseModel):
    ui_id: str = Field(..., description="Task identifier")
    result: str = Field(..., description="Task result message")
    kind: str = Field(..., description="Type of task")
    status: Optional[TaskExecutionStatus] = None
    timestamp: datetime = Field(..., description="ISO timestamp when task completed")
    state: TaskStateMessage


class MessageTaskStarted(BaseModel):
    ui_id: str = Field(..., description="Task identifier")
    kind: str = Field(..., description="Type of task")
    timestamp: datetime = Field(..., description="ISO timestamp when task started")
    state: TaskStateMessage


class MessageTaskFailed(BaseModel):
    ui_id: str = Field(..., description="Task identifier")
    error: str = Field(..., description="Error message")
    kind: str = Field(..., description="Type of task")
    timestamp: datetime = Field(..., description="ISO timestamp when task failed")
    state: TaskStateMessage


class MessageUpdate(
    RootModel[Union[MessageTaskDone, MessageTaskStarted, MessageTaskFailed]]
):
    root: Union[MessageTaskDone, MessageTaskStarted, MessageTaskFailed] = Field(
        ..., description="Union type for all possible WebSocket message updates"
    )


class HistoryResponse(BaseModel):
    history: Optional[Dict[str, TaskHistoryItem]] = Field(
        None, description="Map of task IDs to their history items"
    )