Alex Blank
4c216dd1ca
- Create shared_validators.py with validate_mount_path, validate_files, validate_env_vars, validate_volumes - Refactor config_folders.py to use shared validators - Refactor tool_configs.py to use shared validators - Refactor config_profiles.py to use shared env_vars validator - Reduce ~80 lines of duplicate validation code
338 lines
12 KiB
Python
338 lines
12 KiB
Python
"""Config folder API endpoints."""
|
|
|
|
import logging
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.api.shared_validators import validate_files as _validate_files, validate_mount_path as _validate_mount_path
|
|
from src.auth.dependencies import get_current_user_id, get_db_session
|
|
from src.models.config_folder import ConfigFolder
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/config-folders", tags=["config-folders"])
|
|
|
|
|
|
class ConfigFolderCreate(BaseModel):
|
|
name: str = Field(description="Folder name (unique per user)")
|
|
description: str | None = Field(default=None, description="Optional description")
|
|
mount_path: str = Field(description="Default mount path in container")
|
|
files: dict = Field(default_factory=dict, description="Files as {path: content}")
|
|
|
|
@field_validator("mount_path")
|
|
@classmethod
|
|
def validate_mount_path(cls, v: str) -> str:
|
|
return _validate_mount_path(v)
|
|
|
|
@field_validator("files")
|
|
@classmethod
|
|
def validate_files(cls, v: dict) -> dict:
|
|
return _validate_files(v)
|
|
|
|
|
|
class ConfigFolderUpdate(BaseModel):
|
|
name: str | None = Field(default=None, description="Folder name")
|
|
description: str | None = Field(default=None, description="Optional description")
|
|
mount_path: str | None = Field(default=None, description="Default mount path")
|
|
files: dict | None = Field(default=None, description="Files as {path: content}")
|
|
is_active: bool | None = Field(default=None, description="Active/inactive toggle")
|
|
|
|
@field_validator("mount_path")
|
|
@classmethod
|
|
def validate_mount_path(cls, v: str | None) -> str | None:
|
|
return _validate_mount_path(v)
|
|
|
|
@field_validator("files")
|
|
@classmethod
|
|
def validate_files(cls, v: dict | None) -> dict | None:
|
|
return _validate_files(v)
|
|
|
|
|
|
class ProjectOverrideCreate(BaseModel):
|
|
mount_path: str | None = Field(default=None, description="Override mount path")
|
|
files: dict = Field(default_factory=dict, description="Override files")
|
|
|
|
@field_validator("mount_path")
|
|
@classmethod
|
|
def validate_mount_path(cls, v: str | None) -> str | None:
|
|
return _validate_mount_path(v)
|
|
|
|
|
|
class ConfigFolderResponse(BaseModel):
|
|
id: str
|
|
user_id: str
|
|
name: str
|
|
description: str | None
|
|
mount_path: str
|
|
files: dict
|
|
project_overrides: dict | None
|
|
is_active: bool
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
|
|
@router.get("", summary="List config folders", description="Get all config folders for the current user.")
|
|
async def list_config_folders(
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> dict:
|
|
"""List config folders for the current user."""
|
|
query = select(ConfigFolder).where(ConfigFolder.user_id == user_id)
|
|
result = await session.execute(query)
|
|
folders = result.scalars().all()
|
|
|
|
return {
|
|
"folders": [
|
|
{
|
|
"id": str(f.id),
|
|
"user_id": str(f.user_id),
|
|
"name": f.name,
|
|
"description": f.description,
|
|
"mount_path": f.mount_path,
|
|
"files": f.files,
|
|
"project_overrides": f.project_overrides,
|
|
"is_active": f.is_active,
|
|
"created_at": f.created_at.isoformat() if f.created_at else None,
|
|
"updated_at": f.updated_at.isoformat() if f.updated_at else None,
|
|
}
|
|
for f in folders
|
|
]
|
|
}
|
|
|
|
|
|
@router.post("", summary="Create config folder", description="Create a new config folder.", status_code=status.HTTP_201_CREATED)
|
|
async def create_config_folder(
|
|
data: ConfigFolderCreate,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> dict:
|
|
"""Create a config folder."""
|
|
# Check for duplicate name
|
|
existing = await session.scalar(
|
|
select(ConfigFolder).where(
|
|
ConfigFolder.user_id == user_id,
|
|
ConfigFolder.name == data.name,
|
|
)
|
|
)
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"config folder with name '{data.name}' already exists"
|
|
)
|
|
|
|
folder = ConfigFolder(
|
|
user_id=user_id,
|
|
name=data.name,
|
|
description=data.description,
|
|
mount_path=data.mount_path,
|
|
files=data.files,
|
|
)
|
|
session.add(folder)
|
|
await session.commit()
|
|
await session.refresh(folder)
|
|
|
|
return {
|
|
"id": str(folder.id),
|
|
"user_id": str(folder.user_id),
|
|
"name": folder.name,
|
|
"description": folder.description,
|
|
"mount_path": folder.mount_path,
|
|
"files": folder.files,
|
|
"project_overrides": folder.project_overrides,
|
|
"is_active": folder.is_active,
|
|
"created_at": folder.created_at.isoformat() if folder.created_at else None,
|
|
"updated_at": folder.updated_at.isoformat() if folder.updated_at else None,
|
|
}
|
|
|
|
|
|
@router.put("/{folder_id}", summary="Update config folder", description="Update an existing config folder.")
|
|
async def update_config_folder(
|
|
folder_id: uuid.UUID,
|
|
data: ConfigFolderUpdate,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> dict:
|
|
"""Update a config folder."""
|
|
folder = await session.get(ConfigFolder, folder_id)
|
|
if folder is None or folder.user_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found")
|
|
|
|
if data.name is not None:
|
|
folder.name = data.name
|
|
if data.description is not None:
|
|
folder.description = data.description
|
|
if data.mount_path is not None:
|
|
folder.mount_path = data.mount_path
|
|
if data.files is not None:
|
|
folder.files = data.files
|
|
if data.is_active is not None:
|
|
folder.is_active = data.is_active
|
|
|
|
await session.commit()
|
|
await session.refresh(folder)
|
|
|
|
return {
|
|
"id": str(folder.id),
|
|
"user_id": str(folder.user_id),
|
|
"name": folder.name,
|
|
"description": folder.description,
|
|
"mount_path": folder.mount_path,
|
|
"files": folder.files,
|
|
"project_overrides": folder.project_overrides,
|
|
"is_active": folder.is_active,
|
|
"created_at": folder.created_at.isoformat() if folder.created_at else None,
|
|
"updated_at": folder.updated_at.isoformat() if folder.updated_at else None,
|
|
}
|
|
|
|
|
|
@router.delete("/{folder_id}", summary="Delete config folder", description="Delete a config folder.", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_config_folder(
|
|
folder_id: uuid.UUID,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> None:
|
|
"""Delete a config folder."""
|
|
folder = await session.get(ConfigFolder, folder_id)
|
|
if folder is None or folder.user_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found")
|
|
|
|
await session.delete(folder)
|
|
await session.commit()
|
|
|
|
|
|
class ProjectOverrideWithId(ProjectOverrideCreate):
|
|
project_id: uuid.UUID = Field(description="Project ID for the override")
|
|
|
|
|
|
@router.get("/{folder_id}", summary="Get config folder by ID", description="Get a single config folder by its ID.")
|
|
async def get_config_folder(
|
|
folder_id: uuid.UUID,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> dict:
|
|
"""Get a config folder by ID."""
|
|
folder = await session.get(ConfigFolder, folder_id)
|
|
if folder is None or folder.user_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found")
|
|
|
|
return {
|
|
"id": str(folder.id),
|
|
"user_id": str(folder.user_id),
|
|
"name": folder.name,
|
|
"description": folder.description,
|
|
"mount_path": folder.mount_path,
|
|
"files": folder.files,
|
|
"project_overrides": folder.project_overrides,
|
|
"is_active": folder.is_active,
|
|
"created_at": folder.created_at.isoformat() if folder.created_at else None,
|
|
"updated_at": folder.updated_at.isoformat() if folder.updated_at else None,
|
|
}
|
|
|
|
|
|
@router.post("/{folder_id}/overrides", summary="Add project override", description="Add a project override to a config folder.")
|
|
async def add_project_override(
|
|
folder_id: uuid.UUID,
|
|
data: ProjectOverrideWithId,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> dict:
|
|
"""Add a project override to a config folder."""
|
|
folder = await session.get(ConfigFolder, folder_id)
|
|
if folder is None or folder.user_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found")
|
|
|
|
# Initialize project_overrides if None
|
|
if folder.project_overrides is None:
|
|
folder.project_overrides = {}
|
|
|
|
# Add/update override
|
|
override_data = {}
|
|
if data.mount_path is not None:
|
|
override_data["mount_path"] = data.mount_path
|
|
if data.files is not None:
|
|
override_data["files"] = data.files
|
|
|
|
# Use a copy to trigger SQLAlchemy change detection on JSONB
|
|
current_overrides = dict(folder.project_overrides or {})
|
|
current_overrides[str(data.project_id)] = override_data
|
|
folder.project_overrides = current_overrides
|
|
|
|
await session.commit()
|
|
await session.refresh(folder)
|
|
|
|
return {
|
|
"id": str(folder.id),
|
|
"project_overrides": folder.project_overrides,
|
|
}
|
|
|
|
|
|
@router.put("/{folder_id}/overrides/{project_id}", summary="Update project override", description="Update a project override.")
|
|
async def update_project_override(
|
|
folder_id: uuid.UUID,
|
|
project_id: uuid.UUID,
|
|
data: ProjectOverrideCreate,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> dict:
|
|
"""Update a project override."""
|
|
folder = await session.get(ConfigFolder, folder_id)
|
|
if folder is None or folder.user_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found")
|
|
|
|
# Initialize project_overrides if None
|
|
if folder.project_overrides is None:
|
|
folder.project_overrides = {}
|
|
|
|
# Update override
|
|
current_overrides = dict(folder.project_overrides or {})
|
|
override_data = current_overrides.get(str(project_id), {})
|
|
if data.mount_path is not None:
|
|
override_data["mount_path"] = data.mount_path
|
|
if data.files is not None:
|
|
override_data["files"] = data.files
|
|
|
|
current_overrides[str(project_id)] = override_data
|
|
folder.project_overrides = current_overrides
|
|
|
|
# Mark the field as modified to ensure SQLAlchemy detects the change
|
|
from sqlalchemy.orm.attributes import flag_modified
|
|
flag_modified(folder, "project_overrides")
|
|
|
|
await session.commit()
|
|
await session.refresh(folder)
|
|
|
|
return {
|
|
"id": str(folder.id),
|
|
"project_overrides": folder.project_overrides,
|
|
}
|
|
|
|
|
|
@router.delete("/{folder_id}/overrides/{project_id}", summary="Remove project override", description="Remove a project override.")
|
|
async def remove_project_override(
|
|
folder_id: uuid.UUID,
|
|
project_id: uuid.UUID,
|
|
user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> None:
|
|
"""Remove a project override."""
|
|
folder = await session.get(ConfigFolder, folder_id)
|
|
if folder is None or folder.user_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found")
|
|
|
|
# Remove override if exists
|
|
current_overrides = dict(folder.project_overrides or {})
|
|
if str(project_id) in current_overrides:
|
|
del current_overrides[str(project_id)]
|
|
folder.project_overrides = current_overrides
|
|
await session.commit()
|
|
await session.refresh(folder)
|
|
|
|
return {
|
|
"id": str(folder.id),
|
|
"project_overrides": folder.project_overrides or {},
|
|
}
|