alex
2299bd51ba
When FastAPI parses the request body and model_dump() is called, nested MountItem models are already serialized to plain dicts. The update handler was unconditionally calling model_dump() again, causing AttributeError on dict objects. Fix: Check if mount items are already dicts before calling model_dump(). Fixes: 422 error when updating profiles with mounts.
646 lines
24 KiB
Python
646 lines
24 KiB
Python
"""Config profile API endpoints."""
|
|
|
|
import logging
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from src.auth.dependencies import get_current_user_id, get_db_session
|
|
from src.models.config_profile import ConfigProfile, ConfigProfileInclude
|
|
from src.models.project import Project
|
|
from src.models.tool_type import ToolType
|
|
from src.services.config_profile_resolver import (
|
|
ConfigProfileCycleError,
|
|
check_include_cycle,
|
|
resolve_profile,
|
|
resolved_profile_to_dict,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/config-profiles", tags=["config-profiles"])
|
|
|
|
MAX_PROFILE_SIZE_MB = 10
|
|
MAX_PROFILE_SIZE_BYTES = MAX_PROFILE_SIZE_MB * 1024 * 1024
|
|
|
|
|
|
def _validate_uuid(v: str | None) -> str | None:
|
|
if v is None:
|
|
return v
|
|
try:
|
|
uuid.UUID(v)
|
|
except ValueError:
|
|
raise ValueError(f"Invalid UUID: {v}")
|
|
return v
|
|
|
|
|
|
def _calculate_profile_size(data: dict) -> int:
|
|
"""Calculate approximate serialized size of profile data."""
|
|
total = 0
|
|
for key, value in data.get("env_vars", {}).items():
|
|
total += len(key.encode("utf-8")) + len(str(value).encode("utf-8"))
|
|
for key, value in data.get("runtime_hints", {}).items():
|
|
total += len(key.encode("utf-8")) + len(str(value).encode("utf-8"))
|
|
for mount in data.get("mounts", []):
|
|
total += len(str(mount.get("target", "")).encode("utf-8"))
|
|
total += len(str(mount.get("mode", "")).encode("utf-8"))
|
|
for path, content in mount.get("files", {}).items():
|
|
total += len(path.encode("utf-8")) + len(content.encode("utf-8"))
|
|
for path, content in data.get("files", {}).items():
|
|
total += len(path.encode("utf-8")) + len(content.encode("utf-8"))
|
|
return total
|
|
|
|
|
|
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.keys():
|
|
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}")
|
|
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.keys():
|
|
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:
|
|
if not isinstance(v, dict):
|
|
raise ValueError("env_vars must be a JSON object")
|
|
return v
|
|
|
|
@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}")
|
|
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.keys():
|
|
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:
|
|
raise ValueError(f"Invalid UUID in includes: {item}")
|
|
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
|
|
is_default: bool
|
|
includes: list[dict]
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
|
|
async def _get_profile_with_includes(session: AsyncSession, profile_id: uuid.UUID) -> ConfigProfile | None:
|
|
"""Fetch a profile with includes eagerly loaded."""
|
|
result = await session.execute(
|
|
select(ConfigProfile)
|
|
.where(ConfigProfile.id == profile_id)
|
|
.options(selectinload(ConfigProfile.includes))
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def _check_access(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
project_id: uuid.UUID | None = None,
|
|
tool_type_id: uuid.UUID | None = None,
|
|
) -> None:
|
|
"""Verify user has access to referenced project and tool type."""
|
|
if project_id is not None:
|
|
project = await session.get(Project, project_id)
|
|
if project is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
|
|
# Add ownership check if needed; for now just verify existence
|
|
if tool_type_id is not None:
|
|
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")
|
|
|
|
|
|
def _profile_to_response(profile: ConfigProfile, includes: list[ConfigProfileInclude] | None = None) -> dict:
|
|
return {
|
|
"id": str(profile.id),
|
|
"user_id": str(profile.user_id),
|
|
"name": profile.name,
|
|
"description": profile.description,
|
|
"project_id": str(profile.project_id) if profile.project_id else None,
|
|
"tool_type_id": str(profile.tool_type_id) if profile.tool_type_id else None,
|
|
"env_vars": profile.env_vars or {},
|
|
"runtime_hints": profile.runtime_hints or {},
|
|
"mounts": profile.mounts or [],
|
|
"files": profile.files or {},
|
|
"is_default": profile.is_default,
|
|
"includes": [
|
|
{
|
|
"id": str(inc.id),
|
|
"included_profile_id": str(inc.included_profile_id),
|
|
"order_index": inc.order_index,
|
|
}
|
|
for inc in (includes or profile.includes)
|
|
],
|
|
"created_at": profile.created_at.isoformat() if profile.created_at else None,
|
|
"updated_at": profile.updated_at.isoformat() if profile.updated_at else None,
|
|
}
|
|
|
|
|
|
@router.get("", response_model=list[ConfigProfileResponse])
|
|
async def list_config_profiles(
|
|
project_id: str | None = Query(None, description="Filter by project compatibility"),
|
|
tool_type_id: str | None = Query(None, description="Filter by tool type compatibility"),
|
|
current_user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
):
|
|
"""List config profiles, optionally filtered by compatibility."""
|
|
user_uuid = current_user_id
|
|
query = select(ConfigProfile).where(ConfigProfile.user_id == user_uuid).options(selectinload(ConfigProfile.includes))
|
|
|
|
if project_id or tool_type_id:
|
|
# Compatibility filter: include portable profiles and matching scoped profiles
|
|
project_uuid = uuid.UUID(project_id) if project_id else None
|
|
tool_uuid = uuid.UUID(tool_type_id) if tool_type_id else None
|
|
|
|
from sqlalchemy import or_
|
|
|
|
conditions: list = []
|
|
# Portable profiles (no project, no tool)
|
|
conditions.append(
|
|
(ConfigProfile.project_id.is_(None)) & (ConfigProfile.tool_type_id.is_(None))
|
|
)
|
|
if project_uuid:
|
|
# Profiles matching this project (with or without tool)
|
|
conditions.append(ConfigProfile.project_id == project_uuid)
|
|
if tool_uuid:
|
|
# Profiles matching this tool (with or without project)
|
|
conditions.append(ConfigProfile.tool_type_id == tool_uuid)
|
|
if project_uuid and tool_uuid:
|
|
# Exact match
|
|
conditions.append(
|
|
(ConfigProfile.project_id == project_uuid) & (ConfigProfile.tool_type_id == tool_uuid)
|
|
)
|
|
|
|
query = query.where(or_(*conditions))
|
|
|
|
result = await session.execute(query)
|
|
profiles = result.scalars().all()
|
|
return [_profile_to_response(p) for p in profiles]
|
|
|
|
|
|
@router.post("", response_model=ConfigProfileResponse, status_code=status.HTTP_201_CREATED)
|
|
async def create_config_profile(
|
|
data: ConfigProfileCreate,
|
|
current_user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
):
|
|
"""Create a new config profile."""
|
|
user_uuid = current_user_id
|
|
|
|
# Check for duplicate name
|
|
existing = await session.execute(
|
|
select(ConfigProfile).where(
|
|
ConfigProfile.user_id == user_uuid,
|
|
ConfigProfile.name == data.name,
|
|
).options(selectinload(ConfigProfile.includes))
|
|
)
|
|
if existing.scalar_one_or_none() is not None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"Profile with name '{data.name}' already exists",
|
|
)
|
|
|
|
# Validate references
|
|
project_uuid = uuid.UUID(data.project_id) if data.project_id else None
|
|
tool_uuid = uuid.UUID(data.tool_type_id) if data.tool_type_id else None
|
|
await _check_access(session, user_uuid, project_uuid, tool_uuid)
|
|
|
|
# Check size
|
|
size = _calculate_profile_size(data.model_dump())
|
|
if size > MAX_PROFILE_SIZE_BYTES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail=f"Profile size exceeds {MAX_PROFILE_SIZE_MB}MB limit",
|
|
)
|
|
|
|
profile = ConfigProfile(
|
|
user_id=user_uuid,
|
|
name=data.name,
|
|
description=data.description,
|
|
project_id=project_uuid,
|
|
tool_type_id=tool_uuid,
|
|
env_vars=data.env_vars,
|
|
runtime_hints=data.runtime_hints,
|
|
mounts=[m.model_dump() for m in data.mounts],
|
|
files=data.files,
|
|
is_default=data.is_default,
|
|
)
|
|
session.add(profile)
|
|
await session.commit()
|
|
|
|
# Re-fetch with includes to avoid lazy loading issues
|
|
result = await session.execute(
|
|
select(ConfigProfile)
|
|
.where(ConfigProfile.id == profile.id)
|
|
.options(selectinload(ConfigProfile.includes))
|
|
)
|
|
profile = result.scalar_one()
|
|
|
|
logger.info("Created config profile %s for user %s", profile.id, user_uuid)
|
|
return _profile_to_response(profile)
|
|
|
|
|
|
@router.get("/{profile_id}", response_model=ConfigProfileResponse)
|
|
async def get_config_profile(
|
|
profile_id: str,
|
|
current_user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
):
|
|
"""Get a config profile by ID."""
|
|
profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
|
|
if profile is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
|
|
if profile.user_id != current_user_id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
|
|
return _profile_to_response(profile)
|
|
|
|
|
|
@router.put("/{profile_id}", response_model=ConfigProfileResponse)
|
|
async def update_config_profile(
|
|
profile_id: str,
|
|
data: ConfigProfileUpdate,
|
|
current_user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
):
|
|
"""Update a config profile."""
|
|
profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
|
|
if profile is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
|
|
if profile.user_id != current_user_id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
|
|
# Handle name uniqueness
|
|
if "name" in update_data:
|
|
existing = await session.execute(
|
|
select(ConfigProfile).where(
|
|
ConfigProfile.user_id == profile.user_id,
|
|
ConfigProfile.name == update_data["name"],
|
|
ConfigProfile.id != profile.id,
|
|
)
|
|
)
|
|
if existing.scalar_one_or_none() is not None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"Profile with name '{update_data['name']}' already exists",
|
|
)
|
|
|
|
# Validate references
|
|
project_uuid = (
|
|
uuid.UUID(update_data["project_id"])
|
|
if "project_id" in update_data and update_data["project_id"]
|
|
else (profile.project_id if "project_id" not in update_data else None)
|
|
)
|
|
tool_uuid = (
|
|
uuid.UUID(update_data["tool_type_id"])
|
|
if "tool_type_id" in update_data and update_data["tool_type_id"]
|
|
else (profile.tool_type_id if "tool_type_id" not in update_data else None)
|
|
)
|
|
await _check_access(session, profile.user_id, project_uuid, tool_uuid)
|
|
|
|
# Check size
|
|
current_data = _profile_to_response(profile)
|
|
merged = {**current_data, **update_data}
|
|
size = _calculate_profile_size(merged)
|
|
if size > MAX_PROFILE_SIZE_BYTES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail=f"Profile size exceeds {MAX_PROFILE_SIZE_MB}MB limit",
|
|
)
|
|
|
|
# Apply updates
|
|
for field_name, value in update_data.items():
|
|
if field_name in ("project_id", "tool_type_id"):
|
|
value = uuid.UUID(value) if value else None
|
|
elif field_name == "mounts" and value is not None:
|
|
value = [m.model_dump() if not isinstance(m, dict) else m for m in value]
|
|
setattr(profile, field_name, value)
|
|
|
|
await session.commit()
|
|
|
|
# Re-fetch with includes to avoid lazy loading issues
|
|
result = await session.execute(
|
|
select(ConfigProfile)
|
|
.where(ConfigProfile.id == profile.id)
|
|
.options(selectinload(ConfigProfile.includes))
|
|
)
|
|
profile = result.scalar_one()
|
|
|
|
logger.info("Updated config profile %s", profile.id)
|
|
return _profile_to_response(profile)
|
|
|
|
|
|
@router.delete("/{profile_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_config_profile(
|
|
profile_id: str,
|
|
current_user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
):
|
|
"""Delete a config profile."""
|
|
profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
|
|
if profile is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
|
|
if profile.user_id != current_user_id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
|
|
|
|
await session.delete(profile)
|
|
await session.commit()
|
|
|
|
logger.info("Deleted config profile %s", profile_id)
|
|
return None
|
|
|
|
|
|
@router.put("/{profile_id}/includes", response_model=ConfigProfileResponse)
|
|
async def update_profile_includes(
|
|
profile_id: str,
|
|
data: ConfigProfileIncludeUpdate,
|
|
current_user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
):
|
|
"""Update the ordered includes for a config profile."""
|
|
profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
|
|
if profile is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
|
|
if profile.user_id != current_user_id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
|
|
|
|
# Validate all included profiles exist and belong to the user
|
|
included_uuids = [uuid.UUID(inc_id) for inc_id in data.includes]
|
|
for inc_uuid in included_uuids:
|
|
inc_profile = await session.get(ConfigProfile, inc_uuid)
|
|
if inc_profile is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Included profile not found: {inc_uuid}",
|
|
)
|
|
if inc_profile.user_id != current_user_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Not authorized to include profile: {inc_uuid}",
|
|
)
|
|
if inc_uuid == profile.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Profile cannot include itself",
|
|
)
|
|
|
|
# Check for cycles
|
|
cycle = await check_include_cycle(session, profile.id, None)
|
|
if cycle is None and included_uuids:
|
|
# Check each new include would not create a cycle
|
|
for inc_uuid in included_uuids:
|
|
cycle = await check_include_cycle(session, profile.id, inc_uuid)
|
|
if cycle is not None:
|
|
break
|
|
|
|
if cycle is not None:
|
|
cycle_str = " -> ".join(str(c) for c in cycle)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Include cycle detected: {cycle_str}",
|
|
)
|
|
|
|
# Remove existing includes
|
|
result = await session.execute(
|
|
select(ConfigProfileInclude).where(ConfigProfileInclude.profile_id == profile.id)
|
|
)
|
|
for existing in result.scalars().all():
|
|
await session.delete(existing)
|
|
await session.flush()
|
|
|
|
# Add new includes
|
|
for order_index, inc_uuid in enumerate(included_uuids):
|
|
include = ConfigProfileInclude(
|
|
profile_id=profile.id,
|
|
included_profile_id=inc_uuid,
|
|
order_index=order_index,
|
|
)
|
|
session.add(include)
|
|
await session.flush()
|
|
|
|
await session.commit()
|
|
|
|
# Re-fetch profile (includes loaded separately due to SQLite async issue)
|
|
result = await session.execute(
|
|
select(ConfigProfile).where(ConfigProfile.id == profile.id)
|
|
)
|
|
profile = result.scalar_one()
|
|
|
|
inc_result = await session.execute(
|
|
select(ConfigProfileInclude).where(ConfigProfileInclude.profile_id == profile.id)
|
|
)
|
|
direct_includes = inc_result.scalars().all()
|
|
|
|
logger.info("Updated includes for config profile %s", profile.id)
|
|
return _profile_to_response(profile, list(direct_includes))
|
|
|
|
|
|
@router.get("/{profile_id}/preview")
|
|
async def preview_config_profile(
|
|
profile_id: str,
|
|
current_user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
):
|
|
"""Preview the resolved output of a config profile."""
|
|
profile = await _get_profile_with_includes(session, uuid.UUID(profile_id))
|
|
if profile is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found")
|
|
if profile.user_id != current_user_id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
|
|
|
|
try:
|
|
resolved = await resolve_profile(session, profile.id)
|
|
except ConfigProfileCycleError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(exc),
|
|
)
|
|
|
|
return resolved_profile_to_dict(resolved)
|
|
|
|
|
|
@router.get("/defaults/resolve")
|
|
async def resolve_default_profile(
|
|
project_id: str = Query(..., description="Project ID"),
|
|
tool_type_id: str = Query(..., description="Tool type ID"),
|
|
current_user_id: uuid.UUID = Depends(get_current_user_id),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
):
|
|
"""Resolve the default config profile for a project/tool combination.
|
|
|
|
Selects by specificity:
|
|
1. project+tool explicit default
|
|
2. project explicit default
|
|
3. tool explicit default
|
|
4. global/user explicit default
|
|
5. first created compatible profile
|
|
6. none (returns null)
|
|
"""
|
|
user_uuid = current_user_id
|
|
project_uuid = uuid.UUID(project_id)
|
|
tool_uuid = uuid.UUID(tool_type_id)
|
|
|
|
# Fetch all compatible profiles ordered by created_at
|
|
query = (
|
|
select(ConfigProfile)
|
|
.where(ConfigProfile.user_id == user_uuid)
|
|
.where(
|
|
(ConfigProfile.project_id.is_(None) & ConfigProfile.tool_type_id.is_(None))
|
|
| (ConfigProfile.project_id == project_uuid)
|
|
| (ConfigProfile.tool_type_id == tool_uuid)
|
|
| (
|
|
(ConfigProfile.project_id == project_uuid)
|
|
& (ConfigProfile.tool_type_id == tool_uuid)
|
|
)
|
|
)
|
|
.order_by(ConfigProfile.created_at)
|
|
)
|
|
result = await session.execute(query)
|
|
profiles = result.scalars().all()
|
|
|
|
if not profiles:
|
|
return {"profile_id": None, "profile_name": None}
|
|
|
|
# Check explicit defaults by specificity
|
|
explicit_defaults = [p for p in profiles if p.is_default]
|
|
|
|
# Most specific: project+tool
|
|
for p in explicit_defaults:
|
|
if p.project_id == project_uuid and p.tool_type_id == tool_uuid:
|
|
return {"profile_id": str(p.id), "profile_name": p.name}
|
|
|
|
# Next: project only
|
|
for p in explicit_defaults:
|
|
if p.project_id == project_uuid and p.tool_type_id is None:
|
|
return {"profile_id": str(p.id), "profile_name": p.name}
|
|
|
|
# Next: tool only
|
|
for p in explicit_defaults:
|
|
if p.project_id is None and p.tool_type_id == tool_uuid:
|
|
return {"profile_id": str(p.id), "profile_name": p.name}
|
|
|
|
# Next: global/user (no project, no tool)
|
|
for p in explicit_defaults:
|
|
if p.project_id is None and p.tool_type_id is None:
|
|
return {"profile_id": str(p.id), "profile_name": p.name}
|
|
|
|
# Fall back to first created compatible profile
|
|
first = profiles[0]
|
|
return {"profile_id": str(first.id), "profile_name": first.name}
|