alex
8816ee02ce
Extract inline Pydantic models from 9 API routers into dedicated schema modules under schemas/: - schemas/tool/tool_type.py — ToolTypeCreate, ToolTypeUpdate, etc. - schemas/tool/tool_instance.py — CreateInstanceRequest, StartInstanceRequest - schemas/config/config_profile.py — ConfigProfileCreate, ConfigProfileUpdate, ConfigProfileResponse, DefaultProfilesUpdate, ValidateGitUrlRequest, etc. - schemas/system/health.py — DatabaseHealth, DiskHealth, HealthResponse, etc. - schemas/user/user.py — UserProfileResponse, UserProfileUpdate - schemas/user/user_config.py — UserConfigResponse, UserConfigUpdate - schemas/project/project.py — ProjectCreate, ProjectUpdate, etc. - schemas/project/ssh_key.py — SSHKeyCreate, SSHKeyResponse, etc. - schemas/project/git_repository.py — GitRepositoryCreate, etc. API routers now import from src.schemas.* instead of defining inline. Net change: -548 lines across 16 files. Quality gates: py_compile passed, ruff passed on all 18 files.
281 lines
9.4 KiB
Python
281 lines
9.4 KiB
Python
"""Config profile request/response schemas."""
|
|
|
|
import uuid
|
|
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
from src.api.shared_validators import validate_env_vars as _validate_env_vars
|
|
|
|
|
|
def _validate_uuid(v: str | None) -> str | None:
|
|
if v is None:
|
|
return v
|
|
try:
|
|
uuid.UUID(v)
|
|
except ValueError as exc:
|
|
raise ValueError(f"Invalid UUID: {v}") from exc
|
|
return v
|
|
|
|
|
|
class GitMountMapping(BaseModel):
|
|
source_path: str = Field(
|
|
description="Path within repository (supports glob patterns)"
|
|
)
|
|
target_path: str = Field(description="Absolute path inside container")
|
|
|
|
@field_validator("source_path")
|
|
@classmethod
|
|
def validate_source_path(cls, v: str) -> str:
|
|
if v.startswith("/"):
|
|
raise ValueError("source_path must be relative (no leading /)")
|
|
if ".." in v:
|
|
raise ValueError("source_path cannot contain path traversal (..)")
|
|
return v
|
|
|
|
@field_validator("target_path")
|
|
@classmethod
|
|
def validate_target_path(cls, v: str) -> str:
|
|
if ".." in v:
|
|
raise ValueError("target_path cannot contain path traversal (..)")
|
|
return v
|
|
|
|
|
|
class GitMountItem(BaseModel):
|
|
remote_url: str = Field(description="Git remote URL (HTTPS or SSH)")
|
|
source_path: str | None = Field(
|
|
default=None, description="Path within repository (legacy single mapping)"
|
|
)
|
|
target_path: str | None = Field(
|
|
default=None,
|
|
description="Absolute path inside container (legacy single mapping)",
|
|
)
|
|
branch: str | None = Field(default=None, description="Optional branch or tag name")
|
|
mappings: list[GitMountMapping] | None = Field(
|
|
default=None, description="Multiple source/target mappings from the same repo"
|
|
)
|
|
|
|
@field_validator("remote_url")
|
|
@classmethod
|
|
def validate_remote_url(cls, v: str) -> str:
|
|
if not v.startswith(("http://", "https://", "git@", "ssh://")):
|
|
raise ValueError(
|
|
"remote_url must be a valid git URL (https://, git@, or ssh://)"
|
|
)
|
|
return v
|
|
|
|
@field_validator("source_path")
|
|
@classmethod
|
|
def validate_source_path(cls, v: str | None) -> str | None:
|
|
if v is None:
|
|
return v
|
|
if v.startswith("/"):
|
|
raise ValueError("source_path must be relative (no leading /)")
|
|
if ".." in v:
|
|
raise ValueError("source_path cannot contain path traversal (..)")
|
|
return v
|
|
|
|
@field_validator("target_path")
|
|
@classmethod
|
|
def validate_target_path(cls, v: str | None) -> str | None:
|
|
if v is None:
|
|
return v
|
|
if ".." in v:
|
|
raise ValueError("target_path cannot contain path traversal (..)")
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def check_mappings_or_legacy(self):
|
|
has_legacy = self.source_path is not None and self.target_path is not None
|
|
has_mappings = self.mappings is not None and len(self.mappings) > 0
|
|
if not has_legacy and not has_mappings:
|
|
raise ValueError(
|
|
"Git mount must have either 'mappings' (non-empty array) or both 'source_path' and 'target_path'"
|
|
)
|
|
return self
|
|
|
|
|
|
class MountItem(BaseModel):
|
|
target: str = Field(description="Absolute mount target path")
|
|
mode: str = Field(default="rw", description="Mount mode: ro or rw")
|
|
files: dict = Field(
|
|
default_factory=dict, description="Files as {relative_path: content}"
|
|
)
|
|
|
|
@field_validator("target")
|
|
@classmethod
|
|
def validate_target(cls, v: str) -> str:
|
|
if not v.startswith("/"):
|
|
raise ValueError("Mount target must be absolute (start with /)")
|
|
return v
|
|
|
|
@field_validator("mode")
|
|
@classmethod
|
|
def validate_mode(cls, v: str) -> str:
|
|
if v not in ("ro", "rw"):
|
|
raise ValueError("Mount mode must be 'ro' or 'rw'")
|
|
return v
|
|
|
|
@field_validator("files")
|
|
@classmethod
|
|
def validate_files(cls, v: dict) -> dict:
|
|
for path in v:
|
|
if ".." in path or not path:
|
|
raise ValueError(f"Invalid file path: {path}")
|
|
if path.startswith("/"):
|
|
raise ValueError(
|
|
f"Mount file paths must be relative (got: {path}). "
|
|
f"The mount target defines the absolute container path."
|
|
)
|
|
return v
|
|
|
|
|
|
class ConfigProfileCreate(BaseModel):
|
|
name: str = Field(description="Profile name (unique per user)")
|
|
description: str | None = Field(default=None, description="Optional description")
|
|
project_id: str | None = Field(default=None, description="Optional project ID")
|
|
tool_type_id: str | None = Field(default=None, description="Optional tool type ID")
|
|
env_vars: dict = Field(default_factory=dict, description="Environment variables")
|
|
runtime_hints: dict = Field(default_factory=dict, description="Runtime hints")
|
|
mounts: list[MountItem] = Field(
|
|
default_factory=list, description="Mount definitions"
|
|
)
|
|
files: dict = Field(
|
|
default_factory=dict, description="Files as {relative_path: content}"
|
|
)
|
|
git_mounts: list[GitMountItem] = Field(
|
|
default_factory=list, description="Git repository mounts"
|
|
)
|
|
is_default: bool = Field(
|
|
default=False, description="Whether this is the default profile for its scope"
|
|
)
|
|
|
|
@field_validator("project_id", "tool_type_id")
|
|
@classmethod
|
|
def validate_uuids(cls, v: str | None) -> str | None:
|
|
return _validate_uuid(v)
|
|
|
|
@field_validator("files")
|
|
@classmethod
|
|
def validate_files(cls, v: dict) -> dict:
|
|
for path in v:
|
|
if ".." in path or not path:
|
|
raise ValueError(f"Invalid file path: {path}")
|
|
if path.startswith("/"):
|
|
raise ValueError(
|
|
f"File paths must be relative (got: {path}). "
|
|
f"Use Mounts for absolute container paths."
|
|
)
|
|
return v
|
|
|
|
@field_validator("env_vars")
|
|
@classmethod
|
|
def validate_env_vars(cls, v: dict) -> dict:
|
|
result = _validate_env_vars(v)
|
|
if result is None:
|
|
raise ValueError("env_vars must be a JSON object")
|
|
return result
|
|
|
|
@field_validator("runtime_hints")
|
|
@classmethod
|
|
def validate_runtime_hints(cls, v: dict) -> dict:
|
|
if not isinstance(v, dict):
|
|
raise ValueError("runtime_hints must be a JSON object")
|
|
return v
|
|
|
|
@field_validator("mounts")
|
|
@classmethod
|
|
def validate_mounts(cls, v: list) -> list:
|
|
if not isinstance(v, list):
|
|
raise ValueError("mounts must be a JSON array")
|
|
return v
|
|
|
|
|
|
class ConfigProfileUpdate(BaseModel):
|
|
name: str | None = Field(default=None, description="Profile name")
|
|
description: str | None = Field(default=None, description="Optional description")
|
|
project_id: str | None = Field(default=None, description="Optional project ID")
|
|
tool_type_id: str | None = Field(default=None, description="Optional tool type ID")
|
|
env_vars: dict | None = Field(default=None, description="Environment variables")
|
|
runtime_hints: dict | None = Field(default=None, description="Runtime hints")
|
|
mounts: list[MountItem] | None = Field(
|
|
default=None, description="Mount definitions"
|
|
)
|
|
files: dict | None = Field(
|
|
default=None, description="Files as {relative_path: content}"
|
|
)
|
|
git_mounts: list[GitMountItem] | None = Field(
|
|
default=None, description="Git repository mounts"
|
|
)
|
|
is_default: bool | None = Field(
|
|
default=None, description="Whether this is the default profile"
|
|
)
|
|
|
|
@field_validator("project_id", "tool_type_id")
|
|
@classmethod
|
|
def validate_uuids(cls, v: str | None) -> str | None:
|
|
return _validate_uuid(v)
|
|
|
|
@field_validator("files")
|
|
@classmethod
|
|
def validate_files(cls, v: dict | None) -> dict | None:
|
|
if v is None:
|
|
return v
|
|
for path in v:
|
|
if ".." in path or path.startswith("/") or not path:
|
|
raise ValueError(f"Invalid file path: {path}")
|
|
return v
|
|
|
|
|
|
class ConfigProfileIncludeUpdate(BaseModel):
|
|
includes: list[str] = Field(description="Ordered list of included profile IDs")
|
|
|
|
@field_validator("includes")
|
|
@classmethod
|
|
def validate_includes(cls, v: list) -> list:
|
|
for item in v:
|
|
try:
|
|
uuid.UUID(item)
|
|
except ValueError as exc:
|
|
raise ValueError(f"Invalid UUID in includes: {item}") from exc
|
|
return v
|
|
|
|
|
|
class ConfigProfileResponse(BaseModel):
|
|
id: str
|
|
user_id: str
|
|
name: str
|
|
description: str | None
|
|
project_id: str | None
|
|
tool_type_id: str | None
|
|
env_vars: dict
|
|
runtime_hints: dict
|
|
mounts: list
|
|
files: dict
|
|
git_mounts: list
|
|
is_default: bool
|
|
includes: list[dict]
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
|
|
class DefaultProfilesUpdate(BaseModel):
|
|
default_profiles: dict[str, str] = Field(
|
|
description="Mapping of tool_type_id -> profile_id for default profiles"
|
|
)
|
|
|
|
|
|
class ValidateGitUrlRequest(BaseModel):
|
|
url: str = Field(description="Git remote URL to validate")
|
|
ssh_key_id: str | None = Field(
|
|
default=None, description="Optional SSH key ID for private repos"
|
|
)
|
|
|
|
|
|
class ValidateGitUrlResponse(BaseModel):
|
|
valid: bool
|
|
suggested_url: str | None = None
|
|
branches: list[str] | None = None
|
|
default_branch: str | None = None
|
|
error: str | None = None
|
|
error_code: str | None = None
|