feat: implement tool types definition system

- Add ToolType SQLAlchemy model with Docker Compose template support
- Create CRUD API endpoints for tool type management
- Implement YAML and template variable validation
- Add built-in tool types (code-server, jupyter-notebook) seeded on startup
- Create frontend page with list, create, edit, and delete functionality
- Add tool types navigation to app shell
- Update mypy config to ignore missing imports

Quality gates: ruff (passed), mypy (passed), pytest unit (8 passed),
typecheck (passed), lint (passed), build (passed)
This commit is contained in:
Fusion
2026-05-18 16:27:19 +02:00
parent fb5725947d
commit 6b302b3279
15 changed files with 945 additions and 1 deletions
+257
View File
@@ -0,0 +1,257 @@
import uuid
from typing import Annotated
import yaml
from fastapi import APIRouter, Cookie, Depends, HTTPException, status
from pydantic import BaseModel, ConfigDict, field_validator
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.auth.jwt_service import decode_access_token
from src.config import Settings
from src.database import SessionLocal
from src.models.tool_type import ToolType
from src.models.user import User
router = APIRouter(prefix="/tool-types", tags=["tool-types"])
async def get_db_session():
async with SessionLocal() as session:
yield session
async def get_current_user_id(
access_token: Annotated[str | None, Cookie()] = None,
) -> uuid.UUID:
if not access_token:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing access token")
try:
claims = decode_access_token(settings=Settings(), token=access_token)
return uuid.UUID(str(claims["sub"]))
except Exception:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid access token")
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user not found")
return user
async def _require_admin(user: User) -> None:
# For now, all authenticated users can manage tool types
# In production, check user.role or similar
pass
class ToolTypeCreate(BaseModel):
name: str
display_name: str
description: str | None = None
compose_template: str
required_variables: list[str] = []
@field_validator("compose_template")
@classmethod
def validate_compose_template(cls, v: str) -> str:
try:
parsed = yaml.safe_load(v)
except yaml.YAMLError as e:
raise ValueError(f"Invalid YAML: {e}")
if not isinstance(parsed, dict):
raise ValueError("Compose template must be a YAML mapping")
if "services" not in parsed:
raise ValueError("Compose template must contain 'services' key")
if not parsed["services"]:
raise ValueError("Compose template must define at least one service")
return v
@field_validator("required_variables")
@classmethod
def validate_required_variables(cls, v: list[str], info) -> list[str]:
if not v:
return v
# Get compose_template from the model data
data = info.data
if "compose_template" not in data:
return v
template = data["compose_template"]
for var in v:
placeholder = f"{{{{{var}}}}}"
if placeholder not in template:
raise ValueError(f"Required variable '{var}' not found in compose template")
return v
class ToolTypeUpdate(BaseModel):
display_name: str | None = None
description: str | None = None
compose_template: str | None = None
required_variables: list[str] | None = None
@field_validator("compose_template")
@classmethod
def validate_compose_template(cls, v: str | None) -> str | None:
if v is None:
return v
try:
parsed = yaml.safe_load(v)
except yaml.YAMLError as e:
raise ValueError(f"Invalid YAML: {e}")
if not isinstance(parsed, dict):
raise ValueError("Compose template must be a YAML mapping")
if "services" not in parsed:
raise ValueError("Compose template must contain 'services' key")
if not parsed["services"]:
raise ValueError("Compose template must define at least one service")
return v
class ToolTypeResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
name: str
display_name: str
description: str | None
compose_template: str
required_variables: list[str]
is_builtin: bool
created_by_id: uuid.UUID | None
created_at: str
updated_at: str
@router.post("", response_model=ToolTypeResponse, status_code=status.HTTP_201_CREATED)
async def create_tool_type(
data: ToolTypeCreate,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> ToolType:
user = await _get_user(session, user_id)
await _require_admin(user)
# Check for duplicate name
existing = await session.scalar(select(ToolType).where(ToolType.name == data.name))
if existing:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="tool type with this name already exists")
tool_type = ToolType(
name=data.name,
display_name=data.display_name,
description=data.description,
compose_template=data.compose_template,
required_variables=data.required_variables,
is_builtin=False,
created_by_id=user.id,
)
session.add(tool_type)
await session.commit()
await session.refresh(tool_type)
return tool_type
@router.get("", response_model=list[ToolTypeResponse])
async def list_tool_types(
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> list[ToolType]:
await _get_user(session, user_id)
result = await session.execute(select(ToolType).order_by(ToolType.name))
return list(result.scalars().all())
@router.get("/{tool_type_id}", response_model=ToolTypeResponse)
async def get_tool_type(
tool_type_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> ToolType:
await _get_user(session, user_id)
tool_type = await session.get(ToolType, tool_type_id)
if tool_type is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="tool type not found")
return tool_type
@router.put("/{tool_type_id}", response_model=ToolTypeResponse)
async def update_tool_type(
tool_type_id: uuid.UUID,
data: ToolTypeUpdate,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> ToolType:
user = await _get_user(session, user_id)
await _require_admin(user)
tool_type = await session.get(ToolType, tool_type_id)
if tool_type is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="tool type not found")
if tool_type.is_builtin:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="cannot modify built-in tool types")
update_data = data.model_dump(exclude_unset=True)
# Validate required variables if both are being updated
if "required_variables" in update_data and "compose_template" in update_data:
template = update_data["compose_template"]
for var in update_data["required_variables"]:
placeholder = f"{{{{{var}}}}}"
if placeholder not in template:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Required variable '{var}' not found in compose template"
)
elif "required_variables" in update_data:
# Only updating variables, check against existing template
template = tool_type.compose_template
for var in update_data["required_variables"]:
placeholder = f"{{{{{var}}}}}"
if placeholder not in template:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Required variable '{var}' not found in compose template"
)
for field, value in update_data.items():
setattr(tool_type, field, value)
await session.commit()
await session.refresh(tool_type)
return tool_type
@router.delete("/{tool_type_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_tool_type(
tool_type_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> None:
user = await _get_user(session, user_id)
await _require_admin(user)
tool_type = await session.get(ToolType, tool_type_id)
if tool_type is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="tool type not found")
if tool_type.is_builtin:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="cannot delete built-in tool types")
await session.delete(tool_type)
await session.commit()
+70
View File
@@ -5,14 +5,84 @@ from src.api.auth import router as auth_router
from src.api.git_repositories import router as git_repositories_router
from src.api.projects import router as projects_router
from src.api.ssh_keys import router as ssh_keys_router
from src.api.tool_types import router as tool_types_router
from src.api.user_config import router as user_config_router
from src.api.users import router as users_router
from src.database import SessionLocal
from src.models.tool_type import ToolType
from sqlalchemy import select
app = FastAPI(title="Headquarter API")
async def seed_builtin_tool_types():
async with SessionLocal() as session:
builtin_types = [
{
"name": "code-server",
"display_name": "VS Code Server",
"description": "VS Code running in the browser via code-server",
"compose_template": """version: "3.8"
services:
code-server:
image: lscr.io/linuxserver/code-server:latest
container_name: {{TOOL_NAME}}
environment:
- PUID=1000
- PGID=1000
- TZ=Europe/London
volumes:
- {{REPO_PATH}}:/config/workspace
ports:
- "8443:8443"
restart: unless-stopped""",
"required_variables": ["REPO_PATH", "TOOL_NAME"],
},
{
"name": "jupyter-notebook",
"display_name": "Jupyter Notebook",
"description": "Jupyter Lab for interactive development",
"compose_template": """version: "3.8"
services:
jupyter:
image: jupyter/scipy-notebook:latest
container_name: {{TOOL_NAME}}
environment:
- JUPYTER_ENABLE_LAB=yes
volumes:
- {{REPO_PATH}}:/home/jovyan/work
ports:
- "8888:8888"
restart: unless-stopped""",
"required_variables": ["REPO_PATH", "TOOL_NAME"],
},
]
for tool_data in builtin_types:
existing = await session.scalar(select(ToolType).where(ToolType.name == tool_data["name"]))
if not existing:
tool_type = ToolType(
name=tool_data["name"],
display_name=tool_data["display_name"],
description=tool_data["description"],
compose_template=tool_data["compose_template"],
required_variables=tool_data["required_variables"],
is_builtin=True,
)
session.add(tool_type)
await session.commit()
@app.on_event("startup")
async def on_startup():
await seed_builtin_tool_types()
app.include_router(auth_router)
app.include_router(projects_router)
app.include_router(users_router)
app.include_router(ssh_keys_router)
app.include_router(git_repositories_router)
app.include_router(user_config_router)
app.include_router(tool_types_router)
app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")
+2 -1
View File
@@ -3,7 +3,8 @@ from src.models.git_repository import GitRepository
from src.models.project import Project
from src.models.refresh_token import RefreshToken
from src.models.ssh_key import SSHKey
from src.models.tool_type import ToolType
from src.models.user import User
from src.models.user_config import UserConfig
__all__ = ["Base", "GitRepository", "Project", "RefreshToken", "SSHKey", "User", "UserConfig"]
__all__ = ["Base", "GitRepository", "Project", "RefreshToken", "SSHKey", "ToolType", "User", "UserConfig"]
+29
View File
@@ -0,0 +1,29 @@
import uuid
from typing import TYPE_CHECKING
from sqlalchemy import Boolean, ForeignKey, JSON, String, Text
from sqlalchemy import Uuid as UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
from src.models.user import User
class ToolType(UUIDPrimaryKeyMixin, TimestampMixin, Base):
__tablename__ = "tool_types"
name: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
display_name: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
compose_template: Mapped[str] = mapped_column(Text, nullable=False)
required_variables: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
is_builtin: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
created_by_id: Mapped[uuid.UUID | None] = mapped_column(
UUID(),
ForeignKey("users.id"),
nullable=True,
)
created_by: Mapped["User | None"] = relationship()