alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: extract shared validation and reduce duplication

Browse Source 
- Extract tool_types validation to shared module (validate_compose_yaml, check_port_exposed, validate_required_variables)
- Extract _get_user and _get_owned_project to auth/dependencies.py
- Create useAsyncData hook and apply to 6 pages
- Create extractErrorMessage utility
- TypeScript and build pass
This commit is contained in:
Alex Blank
2026-05-25 14:01:32 +02:00
parent a905cf729e
commit a37a3122f9
19 changed files with 327 additions and 442 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/api/git_repositories.py
+1 -35
View File
@@ -10,7 +10,7 @@ from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.auth.dependencies import get_current_user_id, get_db_session
from src.auth.dependencies import _get_owned_project, _get_user, get_current_user_id, get_db_session
from src.config import Settings
from src.models.git_repository import GitRepository
from src.models.project import Project
@@ -42,40 +42,6 @@ router = APIRouter(prefix="/projects", tags=["git-repositories"])
logger = logging.getLogger(__name__)
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
"""Fetch a user by ID or raise 401 if not found."""
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 _get_owned_project(
project_id: uuid.UUID,
user_id: uuid.UUID,
session: AsyncSession,
) -> Project:
"""Fetch a project and verify ownership.
Args:
project_id: UUID of the project.
user_id: ID of the authenticated user.
session: Database session.
Returns:
The project if found and owned by the user.
Raises:
HTTPException: If project not found or user is not the owner.
"""
project = await session.get(Project, project_id)
if project is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="project not found")
if project.owner_id != user_id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="not project owner")
return project
def _get_repo_path(user_id: uuid.UUID, project_id: uuid.UUID, name: str) -> str:
"""Generate the filesystem path for a repository.
apps/api/src/api/projects.py
+1 -35
View File
@@ -7,7 +7,7 @@ from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.auth.dependencies import get_current_user_id, get_db_session
from src.auth.dependencies import _get_owned_project, _get_user, get_current_user_id, get_db_session
from src.models.git_repository import GitRepository
from src.models.project import Project
from src.models.ssh_key import SSHKey
@@ -16,14 +16,6 @@ from src.models.user import User
router = APIRouter(prefix="/projects", tags=["projects"])
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
"""Fetch a user by ID or raise 401 if not found."""
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
class ProjectCreate(BaseModel):
name: str
description: str | None = None
@@ -132,32 +124,6 @@ async def get_project(
return await _get_owned_project(project_id, user_id, session)
async def _get_owned_project(
project_id: uuid.UUID,
user_id: uuid.UUID,
session: AsyncSession,
) -> Project:
"""Fetch a project and verify ownership.
Args:
project_id: UUID of the project.
user_id: ID of the authenticated user.
session: Database session.
Returns:
The project if found and owned by the user.
Raises:
HTTPException: If project not found or user is not the owner.
"""
project = await session.get(Project, project_id)
if project is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="project not found")
if project.owner_id != user_id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="not project owner")
return project
@router.patch(
"/{project_id}",
response_model=ProjectResponse,
apps/api/src/api/ssh_keys.py
+1 -9
View File
@@ -10,7 +10,7 @@ from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.auth.dependencies import get_current_user_id, get_db_session
from src.auth.dependencies import _get_user, get_current_user_id, get_db_session
from src.config import Settings
from src.models.ssh_key import SSHKey
from src.models.user import User
@@ -18,14 +18,6 @@ from src.models.user import User
router = APIRouter(prefix="/ssh-keys", tags=["ssh-keys"])
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
"""Fetch a user by ID or raise 401 if not found."""
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
def _get_fernet() -> Fernet:
"""Generate a valid Fernet key from the session secret."""
import base64
apps/api/src/api/tool_instances.py
+1 -36
View File
@@ -15,8 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
logger = logging.getLogger(__name__)
from src.auth.dependencies import get_current_user_id
from src.auth.dependencies import get_db_session
from src.auth.dependencies import _get_owned_project, _get_user, get_current_user_id, get_db_session
from src.models.git_repository import GitRepository
from src.models.project import Project
from src.models.ssh_key import SSHKey
@@ -193,40 +192,6 @@ def _modify_compose_file(
compose_file.write_text(yaml.dump(compose_data, default_flow_style=False))
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
"""Fetch a user by ID or raise 404 if not found."""
user = await session.get(User, user_id)
if user is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="user not found"
)
return user
async def _get_owned_project(
project_id: uuid.UUID, user_id: uuid.UUID, session: AsyncSession
) -> Project:
"""Fetch a project and verify ownership.
Args:
project_id: UUID of the project.
user_id: ID of the authenticated user.
session: Database session.
Returns:
The project if found and owned by the user.
Raises:
HTTPException: If project not found or user is not the owner.
"""
project = await session.get(Project, project_id)
if project is None or project.owner_id != user_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="project not found"
)
return project
@router.post(
"/{project_id}/repositories/{repo_id}/instances",
summary="Create tool instance",
apps/api/src/api/tool_types.py
+32 -130
View File
@@ -1,4 +1,3 @@
import re
import uuid
from datetime import datetime
@@ -8,26 +7,19 @@ from pydantic import BaseModel, ConfigDict, field_validator, model_validator
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
def _sanitize_template_vars(template: str) -> str:
"""Replace template variables like {{VAR}} with placeholders to avoid YAML parsing errors."""
return re.sub(r"\{\{[A-Za-z_][A-Za-z0-9_]*\}\}", "__PLACEHOLDER__", template)
from src.auth.dependencies import get_current_user_id, get_db_session
from src.api.tool_types_validation import (
check_port_exposed,
sanitize_template_vars,
validate_compose_yaml,
validate_required_variables,
)
from src.auth.dependencies import _get_user, get_current_user_id, get_db_session
from src.models.tool_type import ToolType
from src.models.user import User
router = APIRouter(prefix="/tool-types", tags=["tool-types"])
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
"""Fetch a user by ID or raise 401 if not found."""
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:
"""Check if user has admin privileges.
@@ -72,24 +64,7 @@ class ToolTypeCreate(BaseModel):
if v is None:
raise ValueError("compose_template is required when definition_type is 'compose'")
# Replace template variables with dummy values before YAML validation
# to avoid YAML parsing errors with {{VAR}} syntax
sanitized = _sanitize_template_vars(v)
try:
parsed = yaml.safe_load(sanitized)
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")
validate_compose_yaml(v)
return v
@field_validator("dockerfile_template")
@@ -156,29 +131,11 @@ class ToolTypeCreate(BaseModel):
# Validate that default_port is exposed in compose template (only if requires_port)
if self.requires_port and self.definition_type == "compose" and self.compose_template:
try:
sanitized = _sanitize_template_vars(self.compose_template)
parsed = yaml.safe_load(sanitized)
except yaml.YAMLError:
parsed = validate_compose_yaml(self.compose_template)
except ValueError:
return self
port_str = str(self.default_port)
port_exposed = False
if isinstance(parsed, dict) and "services" in parsed:
for service_name, service_config in parsed["services"].items():
if isinstance(service_config, dict) and "ports" in service_config:
for port_mapping in service_config["ports"]:
if isinstance(port_mapping, str):
if port_str in port_mapping:
port_exposed = True
break
elif isinstance(port_mapping, int) and port_mapping == self.default_port:
port_exposed = True
break
if port_exposed:
break
if not port_exposed:
if not check_port_exposed(parsed, self.default_port):
raise ValueError(f"Port {self.default_port} is not exposed in the compose template. Add it to the 'ports' section.")
return self
@@ -222,29 +179,13 @@ class ToolTypeUpdate(BaseModel):
def validate_compose_template(cls, v: str | None, info) -> str | None:
if v is None:
return v
data = info.data
definition_type = data.get("definition_type")
if definition_type and definition_type != "compose":
return v
# Replace template variables with dummy values before YAML validation
sanitized = _sanitize_template_vars(v)
try:
parsed = yaml.safe_load(sanitized)
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")
validate_compose_yaml(v)
return v
@field_validator("dockerfile_template")
@@ -442,54 +383,29 @@ async def update_tool_type(
template = update_data.get("compose_template", tool_type.compose_template)
if template:
try:
sanitized = _sanitize_template_vars(template)
parsed = yaml.safe_load(sanitized)
except yaml.YAMLError:
parsed = None
if parsed and isinstance(parsed, dict) and "services" in parsed:
port_str = str(new_port)
port_exposed = False
for service_config in parsed["services"].values():
if isinstance(service_config, dict) and "ports" in service_config:
for port_mapping in service_config["ports"]:
if isinstance(port_mapping, str) and port_str in port_mapping:
port_exposed = True
break
elif isinstance(port_mapping, int) and port_mapping == new_port:
port_exposed = True
break
if port_exposed:
break
if not port_exposed:
parsed = validate_compose_yaml(template)
if not check_port_exposed(parsed, new_port):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Port {new_port} is not exposed in the compose template"
)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e)
)
# Validate required variables for compose definitions
definition_type = update_data.get("definition_type", tool_type.definition_type)
if definition_type == "compose":
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"
)
validate_required_variables(
update_data["compose_template"], update_data["required_variables"]
)
elif "required_variables" in update_data:
template = tool_type.compose_template
if 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"
)
validate_required_variables(template, update_data["required_variables"])
for field, value in update_data.items():
setattr(tool_type, field, value)
@@ -534,16 +450,9 @@ async def validate_tool_type_template(
errors.append("Compose template is required")
else:
try:
sanitized = _sanitize_template_vars(data.compose_template)
parsed = yaml.safe_load(sanitized)
if not isinstance(parsed, dict):
errors.append("Compose template must be a YAML mapping")
elif "services" not in parsed:
errors.append("Compose template must contain 'services' key")
elif not parsed["services"]:
errors.append("Compose template must define at least one service")
except yaml.YAMLError as e:
errors.append(f"Invalid YAML: {e}")
validate_compose_yaml(data.compose_template)
except ValueError as e:
errors.append(str(e))
elif data.definition_type == "dockerfile":
if not data.dockerfile_template:
@@ -592,16 +501,9 @@ async def validate_tool_type(
errors.append("Compose template is empty")
else:
try:
sanitized = _sanitize_template_vars(tool_type.compose_template)
parsed = yaml.safe_load(sanitized)
if not isinstance(parsed, dict):
errors.append("Compose template must be a YAML mapping")
elif "services" not in parsed:
errors.append("Compose template must contain 'services' key")
elif not parsed["services"]:
errors.append("Compose template must define at least one service")
except yaml.YAMLError as e:
errors.append(f"Invalid YAML: {e}")
validate_compose_yaml(tool_type.compose_template)
except ValueError as e:
errors.append(str(e))
elif tool_type.definition_type == "dockerfile":
if not tool_type.dockerfile_template:
apps/api/src/api/tool_types_validation.py
+87
View File
@@ -0,0 +1,87 @@
"""Shared validation utilities for tool types."""
import re
import yaml
from fastapi import HTTPException, status
def sanitize_template_vars(template: str) -> str:
"""Replace template variables like {{VAR}} with placeholders to avoid YAML parsing errors."""
return re.sub(r"\{\{[A-Za-z_][A-Za-z0-9_]*\}\}", "__PLACEHOLDER__", template)
def validate_compose_yaml(template: str) -> dict:
"""Validate and parse a compose template.
Args:
template: Raw compose template string.
Returns:
Parsed YAML dict.
Raises:
ValueError: If YAML is invalid or missing required keys.
"""
sanitized = sanitize_template_vars(template)
try:
parsed = yaml.safe_load(sanitized)
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 parsed
def check_port_exposed(parsed: dict, port: int) -> bool:
"""Check if a port is exposed in a parsed compose template.
Args:
parsed: Parsed compose YAML dict.
port: Port number to check.
Returns:
True if port is exposed in any service.
"""
port_str = str(port)
if not isinstance(parsed, dict) or "services" not in parsed:
return False
for service_config in parsed["services"].values():
if isinstance(service_config, dict) and "ports" in service_config:
for port_mapping in service_config["ports"]:
if isinstance(port_mapping, str) and port_str in port_mapping:
return True
elif isinstance(port_mapping, int) and port_mapping == port:
return True
return False
def validate_required_variables(template: str, variables: list[str]) -> None:
"""Validate that all required variables exist in the template.
Args:
template: Compose template string.
variables: List of required variable names.
Raises:
HTTPException: If any variable is not found in the template.
"""
for var in 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",
)
apps/api/src/api/user_config.py
+1 -9
View File
@@ -8,21 +8,13 @@ from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.auth.dependencies import get_current_user_id, get_db_session
from src.auth.dependencies import _get_user, get_current_user_id, get_db_session
from src.models.user import User
from src.models.user_config import UserConfig
router = APIRouter(prefix="/users/me", tags=["user-config"])
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
"""Fetch a user by ID or raise 401 if not found."""
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 _get_or_create_config(session: AsyncSession, user_id: uuid.UUID) -> UserConfig:
"""Get or create user config record.
apps/api/src/api/users.py
+1 -9
View File
@@ -5,7 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException, UploadFile, status
from pydantic import BaseModel, ConfigDict
from sqlalchemy.ext.asyncio import AsyncSession
from src.auth.dependencies import get_current_user_id, get_db_session
from src.auth.dependencies import _get_user, get_current_user_id, get_db_session
from src.models.user import User
router = APIRouter(prefix="/users", tags=["users"])
@@ -16,14 +16,6 @@ ALLOWED_CONTENT_TYPES = {"image/png", "image/jpeg", "image/jpg"}
MAX_AVATAR_SIZE = 2 * 1024 * 1024 # 2MB
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
"""Fetch a user by ID or raise 401 if not found."""
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
class UserProfileResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
apps/api/src/auth/dependencies.py
+36
View File
@@ -47,3 +47,39 @@ async def get_current_user(
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user not found")
return user
async def _get_user(session: AsyncSession, user_id: uuid.UUID) -> User:
"""Fetch a user by ID or raise 401 if not found."""
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 _get_owned_project(
project_id: uuid.UUID,
user_id: uuid.UUID,
session: AsyncSession,
) -> "Project":
"""Fetch a project and verify ownership.
Args:
project_id: UUID of the project.
user_id: ID of the authenticated user.
session: Database session.
Returns:
The project if found and owned by the user.
Raises:
HTTPException: 404 if project not found, 403 if user is not the owner.
"""
from src.models.project import Project
project = await session.get(Project, project_id)
if project is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="project not found")
if project.owner_id != user_id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="not project owner")
return project