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.
82 lines
2.6 KiB
Python
82 lines
2.6 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.category import Category
|
|
from app.schemas.category import (
|
|
Category as CategorySchema,
|
|
CategoryCreate,
|
|
CategoryUpdate,
|
|
)
|
|
|
|
router = APIRouter(prefix="/categories", tags=["categories"])
|
|
|
|
|
|
@router.get("/", response_model=List[CategorySchema])
|
|
def get_categories(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return db.query(Category).offset(skip).limit(limit).all()
|
|
|
|
|
|
@router.get("/{category_id}", response_model=CategorySchema)
|
|
def get_category(category_id: int, db: Session = Depends(get_db)):
|
|
category = db.query(Category).filter(Category.id == category_id).first()
|
|
if not category:
|
|
raise HTTPException(status_code=404, detail="Category not found")
|
|
return category
|
|
|
|
|
|
@router.post("/", response_model=CategorySchema)
|
|
def create_category(category: CategoryCreate, db: Session = Depends(get_db)):
|
|
existing_category = (
|
|
db.query(Category).filter(Category.name == category.name).first()
|
|
)
|
|
if existing_category:
|
|
raise HTTPException(status_code=400, detail="Category name already exists")
|
|
|
|
db_category = Category(**category.dict())
|
|
db.add(db_category)
|
|
db.commit()
|
|
db.refresh(db_category)
|
|
return db_category
|
|
|
|
|
|
@router.put("/{category_id}", response_model=CategorySchema)
|
|
def update_category(
|
|
category_id: int, category: CategoryUpdate, db: Session = Depends(get_db)
|
|
):
|
|
db_category = db.query(Category).filter(Category.id == category_id).first()
|
|
if not db_category:
|
|
raise HTTPException(status_code=404, detail="Category not found")
|
|
|
|
update_data = category.dict(exclude_unset=True)
|
|
if "name" in update_data:
|
|
existing_category = (
|
|
db.query(Category)
|
|
.filter(Category.name == update_data["name"], Category.id != category_id)
|
|
.first()
|
|
)
|
|
if existing_category:
|
|
raise HTTPException(status_code=400, detail="Category name already exists")
|
|
|
|
for field, value in update_data.items():
|
|
setattr(db_category, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_category)
|
|
return db_category
|
|
|
|
|
|
@router.delete("/{category_id}")
|
|
def delete_category(category_id: int, db: Session = Depends(get_db)):
|
|
db_category = db.query(Category).filter(Category.id == category_id).first()
|
|
if not db_category:
|
|
raise HTTPException(status_code=404, detail="Category not found")
|
|
|
|
db.delete(db_category)
|
|
db.commit()
|
|
return {"message": "Category deleted successfully"}
|