Agent Zero
152 lines
4.9 KiB
Python
152 lines
4.9 KiB
Python
"""StockLocation CRUD endpoints."""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Query
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import require_permission
|
|
from app.db.session import get_async_session
|
|
from app.models import User, StockLocation
|
|
from app.schemas.stock_location import (
|
|
StockLocationCreateRequest,
|
|
StockLocationUpdateRequest,
|
|
StockLocationResponse,
|
|
StockLocationListResponse,
|
|
)
|
|
|
|
router = APIRouter(prefix="/stock-locations", tags=["stock-locations"])
|
|
|
|
|
|
@router.get("", response_model=StockLocationListResponse)
|
|
async def list_stock_locations(
|
|
page: int = Query(1, ge=1),
|
|
size: int = Query(20, ge=1, le=100),
|
|
search: str | None = Query(None, description="Search in name"),
|
|
current_user: User = Depends(require_permission("equipment:read")),
|
|
session: AsyncSession = Depends(get_async_session),
|
|
):
|
|
"""List stock locations with search and pagination."""
|
|
account_id = current_user.account_id
|
|
base_q = select(StockLocation).where(StockLocation.account_id == account_id)
|
|
|
|
if search:
|
|
pattern = f"%{search}%"
|
|
base_q = base_q.where(StockLocation.name.ilike(pattern))
|
|
|
|
count_q = select(func.count()).select_from(base_q.subquery())
|
|
total = (await session.execute(count_q)).scalar() or 0
|
|
|
|
q = base_q.order_by(StockLocation.name).offset((page - 1) * size).limit(size)
|
|
result = await session.execute(q)
|
|
items = result.scalars().all()
|
|
|
|
return StockLocationListResponse(
|
|
items=[StockLocationResponse.model_validate(i) for i in items],
|
|
total=total,
|
|
page=page,
|
|
size=size,
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"", response_model=StockLocationResponse, status_code=status.HTTP_201_CREATED
|
|
)
|
|
async def create_stock_location(
|
|
body: StockLocationCreateRequest,
|
|
current_user: User = Depends(require_permission("equipment:write")),
|
|
session: AsyncSession = Depends(get_async_session),
|
|
):
|
|
"""Create a new stock location."""
|
|
account_id = current_user.account_id
|
|
|
|
# If this is set as default, unset others
|
|
if body.is_default:
|
|
await session.execute(
|
|
select(StockLocation).where(
|
|
StockLocation.account_id == account_id, StockLocation.is_default == True
|
|
) # noqa: E712
|
|
)
|
|
|
|
loc = StockLocation(account_id=account_id, **body.model_dump())
|
|
session.add(loc)
|
|
await session.commit()
|
|
await session.refresh(loc)
|
|
return StockLocationResponse.model_validate(loc)
|
|
|
|
|
|
@router.get("/{location_id}", response_model=StockLocationResponse)
|
|
async def get_stock_location(
|
|
location_id: str,
|
|
current_user: User = Depends(require_permission("equipment:read")),
|
|
session: AsyncSession = Depends(get_async_session),
|
|
):
|
|
"""Get a specific stock location."""
|
|
account_id = current_user.account_id
|
|
result = await session.execute(
|
|
select(StockLocation).where(
|
|
StockLocation.id == location_id,
|
|
StockLocation.account_id == account_id,
|
|
)
|
|
)
|
|
loc = result.scalars().first()
|
|
if not loc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Stock location not found"
|
|
)
|
|
return StockLocationResponse.model_validate(loc)
|
|
|
|
|
|
@router.put("/{location_id}", response_model=StockLocationResponse)
|
|
async def update_stock_location(
|
|
location_id: str,
|
|
body: StockLocationUpdateRequest,
|
|
current_user: User = Depends(require_permission("equipment:write")),
|
|
session: AsyncSession = Depends(get_async_session),
|
|
):
|
|
"""Update a stock location."""
|
|
account_id = current_user.account_id
|
|
result = await session.execute(
|
|
select(StockLocation).where(
|
|
StockLocation.id == location_id,
|
|
StockLocation.account_id == account_id,
|
|
)
|
|
)
|
|
loc = result.scalars().first()
|
|
if not loc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Stock location not found"
|
|
)
|
|
|
|
update_data = body.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(loc, field, value)
|
|
|
|
await session.commit()
|
|
await session.refresh(loc)
|
|
return StockLocationResponse.model_validate(loc)
|
|
|
|
|
|
@router.delete("/{location_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_stock_location(
|
|
location_id: str,
|
|
current_user: User = Depends(require_permission("equipment:delete")),
|
|
session: AsyncSession = Depends(get_async_session),
|
|
):
|
|
"""Delete a stock location."""
|
|
account_id = current_user.account_id
|
|
result = await session.execute(
|
|
select(StockLocation).where(
|
|
StockLocation.id == location_id,
|
|
StockLocation.account_id == account_id,
|
|
)
|
|
)
|
|
loc = result.scalars().first()
|
|
if not loc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Stock location not found"
|
|
)
|
|
|
|
await session.delete(loc)
|
|
await session.commit()
|
|
return None
|