be0ae5f3b3
- Create project structure with FastAPI and SQLAlchemy - Implement database models for items, categories, suppliers, and stock movements - Add CRUD operations for all models - Configure Alembic for database migrations - Create RESTful API endpoints for inventory management - Add health endpoint for system monitoring - Update README with setup and usage instructions generated with BackendIM... (backend.im)
54 lines
2.1 KiB
Python
54 lines
2.1 KiB
Python
from sqlalchemy.orm import Session
|
|
from fastapi import HTTPException, status
|
|
from app.models.category import Category
|
|
from app.schemas.category import CategoryCreate, CategoryUpdate
|
|
|
|
def get_categories(db: Session, skip: int = 0, limit: int = 100, include_inactive: bool = False):
|
|
query = db.query(Category)
|
|
if not include_inactive:
|
|
query = query.filter(Category.is_active == True)
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def get_category(db: Session, category_id: int):
|
|
return db.query(Category).filter(Category.id == category_id).first()
|
|
|
|
def get_category_by_name(db: Session, name: str):
|
|
return db.query(Category).filter(Category.name == name).first()
|
|
|
|
def create_category(db: Session, category: CategoryCreate):
|
|
existing_category = get_category_by_name(db, category.name)
|
|
if existing_category:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Category with this name already exists")
|
|
|
|
db_category = Category(**category.model_dump())
|
|
db.add(db_category)
|
|
db.commit()
|
|
db.refresh(db_category)
|
|
return db_category
|
|
|
|
def update_category(db: Session, category_id: int, category: CategoryUpdate):
|
|
db_category = get_category(db, category_id)
|
|
if not db_category:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Category not found")
|
|
|
|
if category.name and category.name != db_category.name:
|
|
existing_category = get_category_by_name(db, category.name)
|
|
if existing_category:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Category with this name already exists")
|
|
|
|
update_data = category.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_category, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_category)
|
|
return db_category
|
|
|
|
def delete_category(db: Session, category_id: int):
|
|
db_category = get_category(db, category_id)
|
|
if not db_category:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Category not found")
|
|
|
|
db_category.is_active = False
|
|
db.commit()
|
|
return {"message": "Category deleted successfully"} |