RoleForge/app/schemas/domain.py

from datetime import datetime
from typing import Any
from typing import Literal

from pydantic import BaseModel, Field, field_validator


class LintRoleFileError(BaseModel):
    line: int | None = None
    column: int | None = None
    level: str = "error"
    message: str = ""


class LintRoleFileRequest(BaseModel):
    path: str = Field(min_length=1, max_length=255)
    content: str = Field(default="", max_length=600_000)


class LintRoleFileResponse(BaseModel):
    ok: bool
    kind: Literal["yaml", "json"]
    errors: list[LintRoleFileError] = Field(default_factory=list)


class RoleFilePayload(BaseModel):
    path: str = Field(min_length=1, max_length=255)
    content: str = ""


class RoleCreate(BaseModel):
    name: str
    source_type: str = Field(description="galaxy | git | inline")
    source_ref: str = ""
    category_id: str | None = None
    description: str = ""
    content: dict[str, Any] | None = None
    files: list[RoleFilePayload] = Field(default_factory=list)
    visibility: str = Field(default="personal", description="public | team | personal (new roles are always personal)")
    team_id: str | None = Field(default=None, description="If set, role is created for this team (visibility=team).")
    tags: list[str] = Field(default_factory=list, description="Labels for filtering and display (comma-separated in UI).")
    os_families: list[str] = Field(
        default_factory=list,
        description="Target OS families: rhel, debian, alpine, suse, arch, bsd, windows, universal.",
    )


class RoleUpdate(BaseModel):
    name: str
    source_type: str = Field(description="galaxy | git | inline")
    source_ref: str = ""
    category_id: str | None = None
    description: str = ""
    visibility: str = Field(default="personal", description="public | team | personal (new roles are always personal)")
    team_id: str | None = Field(default=None, description="Required when setting visibility to team (or omit to keep current team).")
    tags: list[str] = Field(default_factory=list)
    os_families: list[str] = Field(default_factory=list)


class RoleYamlImportRequest(BaseModel):
    name: str
    tasks_yaml: str
    source_ref: str = "imported.yml"
    category_id: str | None = None
    description: str = ""
    files: list[RoleFilePayload] = Field(default_factory=list)
    visibility: str = Field(default="personal", description="public | team | personal (new roles are always personal)")
    tags: list[str] = Field(default_factory=list)
    os_families: list[str] = Field(default_factory=list)


class RoleFilesReplaceRequest(BaseModel):
    files: list[RoleFilePayload]
    visibility: str = Field(default="personal", description="legacy; server assigns visibility on save")


class RoleForkRequest(BaseModel):
    name: str = Field(min_length=1, max_length=120)
    description: str = ""
    source_type: str = Field(default="inline", description="galaxy | git | inline")
    source_ref: str = ""
    category_id: str | None = None

    @field_validator("category_id", mode="before")
    @classmethod
    def _empty_category_to_none(cls, v: object) -> str | None:
        if v is None or v == "":
            return None
        return str(v)


class RoleImportItem(BaseModel):
    path: str = Field(min_length=1, max_length=255)
    content: str = ""


class RoleImportResult(BaseModel):
    suggested_name: str
    items: list[RoleImportItem]


class RoleCategoryCreate(BaseModel):
    name: str = Field(min_length=2, max_length=80)


class InventoryCreate(BaseModel):
    name: str
    inventory_text: str


class PlaybookCreate(BaseModel):
    name: str
    description: str = ""
    playbook_yaml: str
    inventory_id: str
    role_ids: list[str] = Field(default_factory=list)
    is_shared: bool = False


class PlaybookYamlImportRequest(BaseModel):
    name: str
    playbook_yaml: str
    inventory_id: str
    description: str = ""
    role_ids: list[str] = Field(default_factory=list)
    is_shared: bool = False


class AddRoleToPlaybookRequest(BaseModel):
    role_id: str


class JobLaunchRequest(BaseModel):
    playbook_id: str
    extra_vars: dict[str, Any] = Field(default_factory=dict)
    runtime_mode: Literal["docker", "k8s"] = "docker"


class JobResponse(BaseModel):
    id: str
    status: str
    celery_task_id: str | None
    created_at: datetime


class TestHostSpec(BaseModel):
    name: str
    image: str = "roleforge-backend:latest"
    groups: list[str] = Field(default_factory=list)
    command: str = "tail -f /dev/null"
    privileged: bool = True
    systemd: bool = False
    # OCI platform for molecule/docker_container (e.g. linux/amd64 on ARM hosts for amd64-only images).
    platform: str = ""


class TestLaunchRequest(BaseModel):
    playbook_id: str | None = None
    role_id: str | None = None
    hosts: list[TestHostSpec] = Field(default_factory=list)
    extra_vars: dict[str, Any] = Field(default_factory=dict)
    runtime_mode: Literal["docker", "k8s"] = "docker"


class RunnerStopRequest(BaseModel):
    runtime_mode: Literal["docker", "k8s"]