62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
|
|
from app.core.deps import require_org, require_client_roles, require_write_access, Principal
|
|
from app.models import Batch, Product
|
|
from app.models.enums import ClientRole, QRType
|
|
from app.schemas.common import BatchCreate
|
|
|
|
router = APIRouter(prefix="/batches", tags=["batches"])
|
|
|
|
|
|
def _parse(d):
|
|
if not d:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(d)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _out(b: Batch) -> dict:
|
|
return {"id": str(b.id), "product_id": b.product_id, "batch_number": b.batch_number,
|
|
"mfg_date": b.mfg_date, "expiry_date": b.expiry_date,
|
|
"quantity": b.quantity, "qr_type": b.qr_type, "notes": b.notes,
|
|
"created_at": b.created_at}
|
|
|
|
|
|
@router.get("")
|
|
async def list_batches(p: Principal = Depends(require_org), product_id: Optional[str] = None):
|
|
q = Batch.find(Batch.org_id == str(p.org.id))
|
|
batches = await q.to_list()
|
|
if product_id:
|
|
batches = [b for b in batches if b.product_id == product_id]
|
|
return [_out(b) for b in batches]
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_batch(body: BatchCreate,
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
require_write_access(p)
|
|
product = await Product.get(body.product_id)
|
|
if not product or product.org_id != str(p.org.id):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Product not found")
|
|
bn = body.batch_number or f"{(product.sku or 'B')}{datetime.utcnow():%y%m%d%H%M}"
|
|
batch = Batch(org_id=str(p.org.id), product_id=body.product_id, batch_number=bn,
|
|
mfg_date=_parse(body.mfg_date), expiry_date=_parse(body.expiry_date),
|
|
quantity=max(1, body.quantity), qr_type=body.qr_type, notes=body.notes)
|
|
await batch.insert()
|
|
product.batch_count += 1
|
|
await product.save()
|
|
return _out(batch)
|
|
|
|
|
|
@router.get("/{batch_id}")
|
|
async def get_batch(batch_id: str, p: Principal = Depends(require_org)):
|
|
batch = await Batch.get(batch_id)
|
|
if not batch or batch.org_id != str(p.org.id):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Batch not found")
|
|
return _out(batch)
|