Developer
dd69bd69fc
- Extract git control operations to services/git/control.py with repo validation - Extract git file operations to services/git/files.py with repo validation - Extract repository lifecycle to services/git/repository.py (create, delete, list) - Extract config profile helpers to services/config_profiles.py (cycle detection, duplicate checks, serialization, default profile management) - Slim git_repositories.py from ~1050 to 276 lines - Slim config_profiles.py from ~765 to 299 lines - Both routers now contain only HTTP routing concerns Quality gates: py_compile (pass), file size ≤300 (pass), no subprocess in routers (pass) Refs: repo-restructure Task 3.5
300 lines
9.9 KiB
Python
300 lines
9.9 KiB
Python
"""Config profile business logic."""
|
|
|
|
import logging
|
|
import uuid
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from src.models.config_include import ConfigInclude
|
|
from src.models.config_mount import ConfigMount
|
|
from src.models.config_profile import ConfigProfile
|
|
from src.models.tool_type import ToolType
|
|
from src.models.user_config import UserConfig
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAX_INCLUDES_DEPTH = 10
|
|
|
|
|
|
async def get_owned_profile(
|
|
profile_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
session: AsyncSession,
|
|
) -> ConfigProfile:
|
|
"""Fetch a config profile and verify ownership."""
|
|
profile = await session.get(ConfigProfile, profile_id)
|
|
if profile is None or profile.user_id != user_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="config profile not found",
|
|
)
|
|
return profile
|
|
|
|
|
|
async def _detect_cycle(
|
|
session: AsyncSession,
|
|
profile_id: uuid.UUID,
|
|
visited: set[uuid.UUID] | None = None,
|
|
depth: int = 0,
|
|
) -> bool:
|
|
"""Detect cycles in profile includes using DFS.
|
|
|
|
Returns True if a cycle is detected.
|
|
"""
|
|
if depth > MAX_INCLUDES_DEPTH:
|
|
return True
|
|
|
|
if visited is None:
|
|
visited = set()
|
|
|
|
if profile_id in visited:
|
|
return True
|
|
|
|
visited.add(profile_id)
|
|
|
|
result = await session.execute(
|
|
select(ConfigInclude.included_profile_id).where(
|
|
ConfigInclude.profile_id == profile_id
|
|
)
|
|
)
|
|
included_ids = result.scalars().all()
|
|
|
|
for included_id in included_ids:
|
|
if await _detect_cycle(session, included_id, visited.copy(), depth + 1):
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
async def validate_includes_no_cycle(
|
|
session: AsyncSession,
|
|
profile_id: uuid.UUID,
|
|
new_included_id: uuid.UUID | None = None,
|
|
) -> None:
|
|
"""Validate that adding an include wouldn't create a cycle."""
|
|
if new_included_id and await _detect_cycle(session, new_included_id, {profile_id}):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="adding this include would create a circular reference",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Profile CRUD helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def check_duplicate_name(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
name: str,
|
|
exclude_id: uuid.UUID | None = None,
|
|
) -> None:
|
|
"""Raise 409 if a profile with the given name already exists."""
|
|
query = select(ConfigProfile).where(
|
|
ConfigProfile.user_id == user_id,
|
|
ConfigProfile.name == name,
|
|
)
|
|
if exclude_id:
|
|
query = query.where(ConfigProfile.id != exclude_id)
|
|
existing = await session.scalar(query)
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"config profile with name '{name}' already exists",
|
|
)
|
|
|
|
|
|
def profile_to_dict(profile: ConfigProfile) -> dict:
|
|
"""Serialize a ConfigProfile to a dict."""
|
|
return {
|
|
"id": str(profile.id),
|
|
"user_id": str(profile.user_id),
|
|
"name": profile.name,
|
|
"description": profile.description,
|
|
"created_at": profile.created_at.isoformat() if profile.created_at else None,
|
|
"updated_at": profile.updated_at.isoformat() if profile.updated_at else None,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Include helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def check_duplicate_include(
|
|
session: AsyncSession,
|
|
profile_id: uuid.UUID,
|
|
included_profile_id: uuid.UUID,
|
|
) -> None:
|
|
"""Raise 409 if the include already exists."""
|
|
existing = await session.scalar(
|
|
select(ConfigInclude).where(
|
|
ConfigInclude.profile_id == profile_id,
|
|
ConfigInclude.included_profile_id == included_profile_id,
|
|
)
|
|
)
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="this include already exists",
|
|
)
|
|
|
|
|
|
def include_to_dict(inc: ConfigInclude, included_name: str | None) -> dict:
|
|
"""Serialize a ConfigInclude to a dict."""
|
|
return {
|
|
"id": str(inc.id),
|
|
"profile_id": str(inc.profile_id),
|
|
"included_profile_id": str(inc.included_profile_id),
|
|
"included_profile_name": included_name,
|
|
"order_index": inc.order_index,
|
|
"created_at": inc.created_at.isoformat() if inc.created_at else None,
|
|
"updated_at": inc.updated_at.isoformat() if inc.updated_at else None,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mount helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def check_duplicate_mount_path(
|
|
session: AsyncSession,
|
|
profile_id: uuid.UUID,
|
|
target_path: str,
|
|
exclude_id: uuid.UUID | None = None,
|
|
) -> None:
|
|
"""Raise 409 if a mount with the given path already exists."""
|
|
query = select(ConfigMount).where(
|
|
ConfigMount.profile_id == profile_id,
|
|
ConfigMount.target_path == target_path,
|
|
)
|
|
if exclude_id:
|
|
query = query.where(ConfigMount.id != exclude_id)
|
|
existing = await session.scalar(query)
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"mount with path '{target_path}' already exists",
|
|
)
|
|
|
|
|
|
def mount_to_dict(mount: ConfigMount) -> dict:
|
|
"""Serialize a ConfigMount to a dict."""
|
|
return {
|
|
"id": str(mount.id),
|
|
"profile_id": str(mount.profile_id),
|
|
"target_path": mount.target_path,
|
|
"files": mount.files,
|
|
"mode": mount.mode,
|
|
"order_index": mount.order_index,
|
|
"created_at": mount.created_at.isoformat() if mount.created_at else None,
|
|
"updated_at": mount.updated_at.isoformat() if mount.updated_at else None,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Default profile helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def get_or_create_user_config(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
) -> UserConfig:
|
|
"""Get existing user config or create a new one."""
|
|
result = await session.execute(select(UserConfig).where(UserConfig.user_id == user_id))
|
|
user_config = result.scalar_one_or_none()
|
|
if user_config is None:
|
|
user_config = UserConfig(user_id=user_id, config={})
|
|
session.add(user_config)
|
|
return user_config
|
|
|
|
|
|
async def validate_default_profiles(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
default_profiles: dict[str, str],
|
|
) -> None:
|
|
"""Validate that all profile IDs in default_profiles belong to the user."""
|
|
for tool_type_id, profile_id_str in default_profiles.items():
|
|
profile = await session.get(ConfigProfile, uuid.UUID(profile_id_str))
|
|
if profile is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"profile {profile_id_str} not found")
|
|
if profile.user_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"profile {profile_id_str} does not belong to user")
|
|
|
|
|
|
async def get_default_profiles(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
) -> dict:
|
|
"""Get default profiles for a user."""
|
|
result = await session.execute(select(UserConfig).where(UserConfig.user_id == user_id))
|
|
user_config = result.scalar_one_or_none()
|
|
return {"default_profiles": user_config.default_profiles if user_config else {}}
|
|
|
|
|
|
async def set_default_profiles(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
default_profiles: dict[str, str],
|
|
) -> dict:
|
|
"""Set default profiles for a user."""
|
|
user_config = await get_or_create_user_config(session, user_id)
|
|
await validate_default_profiles(session, user_id, default_profiles)
|
|
user_config.config = {**user_config.config, "default_profiles": default_profiles}
|
|
await session.commit()
|
|
await session.refresh(user_config)
|
|
return {"default_profiles": user_config.default_profiles}
|
|
|
|
|
|
async def get_default_profile_for_tool_type(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
tool_type_id: str,
|
|
) -> dict:
|
|
"""Get default profile for a specific tool type."""
|
|
result = await session.execute(select(UserConfig).where(UserConfig.user_id == user_id))
|
|
user_config = result.scalar_one_or_none()
|
|
profile_id = user_config.default_profiles.get(tool_type_id) if user_config else None
|
|
return {"tool_type_id": tool_type_id, "profile_id": profile_id}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Include list helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def list_includes_for_profile(
|
|
session: AsyncSession,
|
|
profile_id: uuid.UUID,
|
|
) -> dict:
|
|
"""List all includes for a profile."""
|
|
result = await session.execute(
|
|
select(ConfigInclude)
|
|
.where(ConfigInclude.profile_id == profile_id)
|
|
.order_by(ConfigInclude.order_index)
|
|
)
|
|
includes_data = []
|
|
for inc in result.scalars().all():
|
|
included_profile = await session.get(ConfigProfile, inc.included_profile_id)
|
|
includes_data.append(include_to_dict(inc, included_profile.name if included_profile else None))
|
|
return {"includes": includes_data}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mount list helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def list_mounts_for_profile(
|
|
session: AsyncSession,
|
|
profile_id: uuid.UUID,
|
|
) -> dict:
|
|
"""List all mounts for a profile."""
|
|
result = await session.execute(
|
|
select(ConfigMount)
|
|
.where(ConfigMount.profile_id == profile_id)
|
|
.order_by(ConfigMount.order_index)
|
|
)
|
|
return {"mounts": [mount_to_dict(m) for m in result.scalars().all()]}
|