108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from typing import Optional, List
|
|
|
|
from app.database import get_db
|
|
from app.models.location import Location
|
|
from app.schemas.location import LocationCreate, LocationUpdate, LocationResponse
|
|
from app.routers.auth import get_current_session
|
|
from app.models.settings import Settings
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[LocationResponse])
|
|
async def get_locations(
|
|
category: Optional[str] = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Settings = Depends(get_current_session)
|
|
):
|
|
"""Get all locations with optional category filter."""
|
|
query = select(Location)
|
|
|
|
if category:
|
|
query = query.where(Location.category == category)
|
|
|
|
query = query.order_by(Location.name.asc())
|
|
|
|
result = await db.execute(query)
|
|
locations = result.scalars().all()
|
|
|
|
return locations
|
|
|
|
|
|
@router.post("/", response_model=LocationResponse, status_code=201)
|
|
async def create_location(
|
|
location: LocationCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Settings = Depends(get_current_session)
|
|
):
|
|
"""Create a new location."""
|
|
new_location = Location(**location.model_dump())
|
|
db.add(new_location)
|
|
await db.commit()
|
|
await db.refresh(new_location)
|
|
|
|
return new_location
|
|
|
|
|
|
@router.get("/{location_id}", response_model=LocationResponse)
|
|
async def get_location(
|
|
location_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Settings = Depends(get_current_session)
|
|
):
|
|
"""Get a specific location by ID."""
|
|
result = await db.execute(select(Location).where(Location.id == location_id))
|
|
location = result.scalar_one_or_none()
|
|
|
|
if not location:
|
|
raise HTTPException(status_code=404, detail="Location not found")
|
|
|
|
return location
|
|
|
|
|
|
@router.put("/{location_id}", response_model=LocationResponse)
|
|
async def update_location(
|
|
location_id: int,
|
|
location_update: LocationUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Settings = Depends(get_current_session)
|
|
):
|
|
"""Update a location."""
|
|
result = await db.execute(select(Location).where(Location.id == location_id))
|
|
location = result.scalar_one_or_none()
|
|
|
|
if not location:
|
|
raise HTTPException(status_code=404, detail="Location not found")
|
|
|
|
update_data = location_update.model_dump(exclude_unset=True)
|
|
|
|
for key, value in update_data.items():
|
|
setattr(location, key, value)
|
|
|
|
await db.commit()
|
|
await db.refresh(location)
|
|
|
|
return location
|
|
|
|
|
|
@router.delete("/{location_id}", status_code=204)
|
|
async def delete_location(
|
|
location_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: Settings = Depends(get_current_session)
|
|
):
|
|
"""Delete a location."""
|
|
result = await db.execute(select(Location).where(Location.id == location_id))
|
|
location = result.scalar_one_or_none()
|
|
|
|
if not location:
|
|
raise HTTPException(status_code=404, detail="Location not found")
|
|
|
|
await db.delete(location)
|
|
await db.commit()
|
|
|
|
return None
|