89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
"""The suppliers endpoint
|
|
"""
|
|
from fastapi import APIRouter, HTTPException
|
|
from sqlmodel import select
|
|
from app.api.deps import SessionDep, exists
|
|
from app.schemas.models import Supplier
|
|
from app.schemas.schemas import SupplierCreate, SupplierUpdate
|
|
from pydantic import ValidationError
|
|
from typing import Sequence, List, Optional
|
|
|
|
|
|
router = APIRouter(prefix="/suppliers", tags=["suppliers"])
|
|
|
|
|
|
@router.get("/", response_model=List[Supplier])
|
|
def fetch_suppliers(session: SessionDep) -> Sequence[Supplier]:
|
|
"""Fetch supplier list
|
|
"""
|
|
suppliers = session.exec(select(Supplier)).all()
|
|
return suppliers
|
|
|
|
|
|
@router.post("/", response_model=SupplierCreate)
|
|
def create_supplier(supplier_data: SupplierCreate, session: SessionDep) -> Supplier:
|
|
"""Create a supplier
|
|
"""
|
|
existing = exists(session, Supplier, tin_number=supplier_data.tin_number)
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Supplier with this tin_number already exists")
|
|
|
|
try:
|
|
supplier = Supplier.model_validate(supplier_data)
|
|
except ValidationError as e:
|
|
raise HTTPException(status_code=400, detail=e.errors())
|
|
|
|
session.add(supplier)
|
|
session.commit()
|
|
session.refresh(supplier)
|
|
return supplier
|
|
|
|
|
|
@router.get("/{supplier_id}", response_model=Supplier)
|
|
def get_supplier(supplier_id: int, session: SessionDep) -> Optional[Supplier]:
|
|
"""Returns a supplier by its ID
|
|
"""
|
|
stmt = select(Supplier).where(Supplier.id == supplier_id)
|
|
result: Optional[Supplier] = session.exec(stmt).first()
|
|
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Supplier not found")
|
|
|
|
return result
|
|
|
|
|
|
@router.patch("/{supplier_id}", response_model=SupplierUpdate)
|
|
def update_supplier(
|
|
supplier_id: int,
|
|
supplier_data: SupplierUpdate,
|
|
session: SessionDep
|
|
) -> Optional[Supplier]:
|
|
"""Updates a supplier's details
|
|
"""
|
|
supplier = session.get(Supplier, supplier_id)
|
|
if not supplier:
|
|
raise HTTPException(status_code=404, detail="Supplier not found")
|
|
|
|
updated_fields = supplier_data.model_dump(exclude_unset=True)
|
|
|
|
for key, value in updated_fields.items():
|
|
setattr(supplier, key, value)
|
|
|
|
session.add(supplier)
|
|
session.commit()
|
|
session.refresh(supplier)
|
|
return supplier
|
|
|
|
|
|
@router.delete("/{supplier_id}", status_code=204)
|
|
def delete_supplier(supplier_id: int, session: SessionDep):
|
|
"""Deletes a supplier
|
|
"""
|
|
supplier = session.get(Supplier, supplier_id)
|
|
|
|
if not supplier:
|
|
raise HTTPException(status_code=404, detail="Supplier not found")
|
|
|
|
session.delete(supplier)
|
|
session.commit()
|