Fusion
123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.auth.dependencies import get_current_active_user
|
|
from app.db import get_db_session
|
|
from app.models.config import Config
|
|
from app.models.project import Project
|
|
from app.models.tool_instance import ToolInstance
|
|
from app.models.user import User
|
|
from app.schemas.config import ConfigCreate, ConfigRead, ConfigUpdate
|
|
|
|
router = APIRouter(tags=["configs"])
|
|
|
|
|
|
async def _verify_config_ownership(
|
|
config_obj: Config, user: User, session: AsyncSession
|
|
) -> None:
|
|
if config_obj.scope_type == "user":
|
|
if config_obj.scope_id != user.id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
|
|
elif config_obj.scope_type == "project":
|
|
project = await session.get(Project, config_obj.scope_id)
|
|
if not project or project.owner_id != user.id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
|
|
elif config_obj.scope_type == "tool_instance":
|
|
ti = await session.get(ToolInstance, config_obj.scope_id)
|
|
if not ti:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
|
|
project = await session.get(Project, ti.project_id)
|
|
if not project or project.owner_id != user.id:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
|
|
elif config_obj.scope_type == "global":
|
|
pass
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid scope_type")
|
|
|
|
|
|
@router.post("/configs", response_model=ConfigRead, status_code=status.HTTP_201_CREATED)
|
|
async def create_config(
|
|
config_in: ConfigCreate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> Config:
|
|
cfg = Config(**config_in.model_dump())
|
|
await _verify_config_ownership(cfg, current_user, session)
|
|
session.add(cfg)
|
|
await session.commit()
|
|
await session.refresh(cfg)
|
|
return cfg
|
|
|
|
|
|
@router.get("/configs", response_model=list[ConfigRead])
|
|
async def list_configs(
|
|
scope_type: str | None = None,
|
|
scope_id: UUID | None = None,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> list[Config]:
|
|
stmt = select(Config)
|
|
if scope_type:
|
|
stmt = stmt.where(Config.scope_type == scope_type)
|
|
if scope_id:
|
|
stmt = stmt.where(Config.scope_id == scope_id)
|
|
result = await session.execute(stmt)
|
|
configs = list(result.scalars().all())
|
|
allowed = []
|
|
for cfg in configs:
|
|
try:
|
|
await _verify_config_ownership(cfg, current_user, session)
|
|
allowed.append(cfg)
|
|
except HTTPException:
|
|
pass
|
|
return allowed
|
|
|
|
|
|
@router.get("/configs/{config_id}", response_model=ConfigRead)
|
|
async def get_config(
|
|
config_id: UUID,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> Config:
|
|
cfg = await session.get(Config, config_id)
|
|
if not cfg:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Config not found")
|
|
await _verify_config_ownership(cfg, current_user, session)
|
|
return cfg
|
|
|
|
|
|
@router.put("/configs/{config_id}", response_model=ConfigRead)
|
|
async def update_config(
|
|
config_id: UUID,
|
|
config_in: ConfigUpdate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> Config:
|
|
cfg = await session.get(Config, config_id)
|
|
if not cfg:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Config not found")
|
|
await _verify_config_ownership(cfg, current_user, session)
|
|
update_data = config_in.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(cfg, field, value)
|
|
await session.commit()
|
|
await session.refresh(cfg)
|
|
return cfg
|
|
|
|
|
|
@router.delete("/configs/{config_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_config(
|
|
config_id: UUID,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> None:
|
|
cfg = await session.get(Config, config_id)
|
|
if not cfg:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Config not found")
|
|
await _verify_config_ownership(cfg, current_user, session)
|
|
await session.delete(cfg)
|
|
await session.commit()
|