83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.dependencies import get_current_user, get_db
|
|
from app.models.object import Object
|
|
from app.models.zone import Zone
|
|
from app.models.user import User
|
|
from app.schemas.zone import ZoneCreate, ZoneRead, ZoneUpdate
|
|
|
|
router = APIRouter(prefix="/api/v1/objects/{obj_id}/zones", tags=["zones"])
|
|
|
|
|
|
async def _get_object(db: AsyncSession, obj_id: int) -> Object:
|
|
obj = await db.get(Object, obj_id)
|
|
if obj is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Object not found")
|
|
return obj
|
|
|
|
|
|
async def _get_zone(db: AsyncSession, obj_id: int, zone_id: int) -> Zone:
|
|
result = await db.execute(
|
|
select(Zone).where(Zone.id == zone_id, Zone.object_id == obj_id)
|
|
)
|
|
zone = result.scalar_one_or_none()
|
|
if zone is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Zone not found")
|
|
return zone
|
|
|
|
|
|
@router.get("", response_model=list[ZoneRead])
|
|
async def list_zones(
|
|
obj_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(get_current_user),
|
|
):
|
|
await _get_object(db, obj_id)
|
|
result = await db.execute(
|
|
select(Zone).where(Zone.object_id == obj_id).order_by(Zone.name)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("", response_model=ZoneRead, status_code=status.HTTP_201_CREATED)
|
|
async def create_zone(
|
|
obj_id: int,
|
|
body: ZoneCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(get_current_user),
|
|
):
|
|
await _get_object(db, obj_id)
|
|
zone = Zone(object_id=obj_id, **body.model_dump())
|
|
db.add(zone)
|
|
await db.flush()
|
|
await db.refresh(zone)
|
|
return zone
|
|
|
|
|
|
@router.patch("/{zone_id}", response_model=ZoneRead)
|
|
async def update_zone(
|
|
obj_id: int,
|
|
zone_id: int,
|
|
body: ZoneUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(get_current_user),
|
|
):
|
|
zone = await _get_zone(db, obj_id, zone_id)
|
|
for field, value in body.model_dump(exclude_none=True).items():
|
|
setattr(zone, field, value)
|
|
await db.flush()
|
|
await db.refresh(zone)
|
|
return zone
|
|
|
|
|
|
@router.delete("/{zone_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_zone(
|
|
obj_id: int,
|
|
zone_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(get_current_user),
|
|
):
|
|
zone = await _get_zone(db, obj_id, zone_id)
|
|
await db.delete(zone)
|