447799c9a0
Features implemented: - Product management with CRUD operations - Category and supplier management - Stock movement tracking with automatic updates - Inventory reports and analytics - SQLite database with Alembic migrations - Health monitoring endpoints - CORS configuration for API access - Comprehensive API documentation - Code quality with Ruff linting and formatting The system provides a complete backend solution for small business inventory management with proper database relationships, stock tracking, and reporting capabilities.
67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
from typing import List
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from app.db.session import get_db
|
|
from app.models.supplier import Supplier
|
|
from app.schemas.supplier import (
|
|
Supplier as SupplierSchema,
|
|
SupplierCreate,
|
|
SupplierUpdate,
|
|
)
|
|
|
|
router = APIRouter(prefix="/suppliers", tags=["suppliers"])
|
|
|
|
|
|
@router.get("/", response_model=List[SupplierSchema])
|
|
def get_suppliers(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return db.query(Supplier).offset(skip).limit(limit).all()
|
|
|
|
|
|
@router.get("/{supplier_id}", response_model=SupplierSchema)
|
|
def get_supplier(supplier_id: int, db: Session = Depends(get_db)):
|
|
supplier = db.query(Supplier).filter(Supplier.id == supplier_id).first()
|
|
if not supplier:
|
|
raise HTTPException(status_code=404, detail="Supplier not found")
|
|
return supplier
|
|
|
|
|
|
@router.post("/", response_model=SupplierSchema)
|
|
def create_supplier(supplier: SupplierCreate, db: Session = Depends(get_db)):
|
|
db_supplier = Supplier(**supplier.dict())
|
|
db.add(db_supplier)
|
|
db.commit()
|
|
db.refresh(db_supplier)
|
|
return db_supplier
|
|
|
|
|
|
@router.put("/{supplier_id}", response_model=SupplierSchema)
|
|
def update_supplier(
|
|
supplier_id: int, supplier: SupplierUpdate, db: Session = Depends(get_db)
|
|
):
|
|
db_supplier = db.query(Supplier).filter(Supplier.id == supplier_id).first()
|
|
if not db_supplier:
|
|
raise HTTPException(status_code=404, detail="Supplier not found")
|
|
|
|
update_data = supplier.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_supplier, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_supplier)
|
|
return db_supplier
|
|
|
|
|
|
@router.delete("/{supplier_id}")
|
|
def delete_supplier(supplier_id: int, db: Session = Depends(get_db)):
|
|
db_supplier = db.query(Supplier).filter(Supplier.id == supplier_id).first()
|
|
if not db_supplier:
|
|
raise HTTPException(status_code=404, detail="Supplier not found")
|
|
|
|
db.delete(db_supplier)
|
|
db.commit()
|
|
return {"message": "Supplier deleted successfully"}
|