Fusion
104 lines
4.2 KiB
Python
104 lines
4.2 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.access_route import AccessRoute
|
|
from app.models.project import Project
|
|
from app.models.tool_instance import ToolInstance
|
|
from app.models.user import User
|
|
from app.schemas.access_route import AccessRouteCreate, AccessRouteRead, AccessRouteUpdate
|
|
|
|
router = APIRouter(tags=["access-routes"])
|
|
|
|
|
|
async def _verify_tool_instance_ownership(
|
|
instance_id: UUID, user: User, session: AsyncSession
|
|
) -> None:
|
|
ti = await session.get(ToolInstance, instance_id)
|
|
if not ti:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tool instance not found")
|
|
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")
|
|
|
|
|
|
@router.post("/tool-instances/{instance_id}/access-routes", response_model=AccessRouteRead, status_code=status.HTTP_201_CREATED) # noqa: E501
|
|
async def create_access_route(
|
|
instance_id: UUID,
|
|
ar_in: AccessRouteCreate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> AccessRoute:
|
|
await _verify_tool_instance_ownership(instance_id, current_user, session)
|
|
ar = AccessRoute(**ar_in.model_dump(), tool_instance_id=instance_id)
|
|
session.add(ar)
|
|
await session.commit()
|
|
await session.refresh(ar)
|
|
return ar
|
|
|
|
|
|
@router.get("/tool-instances/{instance_id}/access-routes", response_model=list[AccessRouteRead])
|
|
async def list_access_routes(
|
|
instance_id: UUID,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> list[AccessRoute]:
|
|
await _verify_tool_instance_ownership(instance_id, current_user, session)
|
|
result = await session.execute(
|
|
select(AccessRoute).where(AccessRoute.tool_instance_id == instance_id)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
@router.get("/tool-instances/{instance_id}/access-routes/{route_id}", response_model=AccessRouteRead) # noqa: E501
|
|
async def get_access_route(
|
|
instance_id: UUID,
|
|
route_id: UUID,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> AccessRoute:
|
|
await _verify_tool_instance_ownership(instance_id, current_user, session)
|
|
ar = await session.get(AccessRoute, route_id)
|
|
if not ar or ar.tool_instance_id != instance_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Access route not found")
|
|
return ar
|
|
|
|
|
|
@router.put("/tool-instances/{instance_id}/access-routes/{route_id}", response_model=AccessRouteRead) # noqa: E501
|
|
async def update_access_route(
|
|
instance_id: UUID,
|
|
route_id: UUID,
|
|
ar_in: AccessRouteUpdate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> AccessRoute:
|
|
await _verify_tool_instance_ownership(instance_id, current_user, session)
|
|
ar = await session.get(AccessRoute, route_id)
|
|
if not ar or ar.tool_instance_id != instance_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Access route not found")
|
|
update_data = ar_in.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(ar, field, value)
|
|
await session.commit()
|
|
await session.refresh(ar)
|
|
return ar
|
|
|
|
|
|
@router.delete("/tool-instances/{instance_id}/access-routes/{route_id}", status_code=status.HTTP_204_NO_CONTENT) # noqa: E501
|
|
async def delete_access_route(
|
|
instance_id: UUID,
|
|
route_id: UUID,
|
|
current_user: User = Depends(get_current_active_user),
|
|
session: AsyncSession = Depends(get_db_session),
|
|
) -> None:
|
|
await _verify_tool_instance_ownership(instance_id, current_user, session)
|
|
ar = await session.get(AccessRoute, route_id)
|
|
if not ar or ar.tool_instance_id != instance_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Access route not found")
|
|
await session.delete(ar)
|
|
await session.commit()
|