from typing import List, Optional from sqlalchemy.orm import Session from app.models.category import Category from app.schemas.category import CategoryCreate, CategoryUpdate def get_category(db: Session, category_id: int) -> Optional[Category]: """ Get a category by ID """ return db.query(Category).filter(Category.id == category_id).first() def get_category_by_user( db: Session, category_id: int, user_id: int ) -> Optional[Category]: """ Get a category by ID for a specific user """ return db.query(Category).filter( Category.id == category_id, Category.owner_id == user_id ).first() def get_categories( db: Session, user_id: int, skip: int = 0, limit: int = 100 ) -> List[Category]: """ Get all categories for a user """ return db.query(Category).filter( Category.owner_id == user_id ).offset(skip).limit(limit).all() def create_category( db: Session, category_in: CategoryCreate, user_id: int ) -> Category: """ Create a new category for a user """ db_category = Category( **category_in.dict(), owner_id=user_id, ) db.add(db_category) db.commit() db.refresh(db_category) return db_category def update_category( db: Session, category_id: int, category_in: CategoryUpdate, user_id: int ) -> Optional[Category]: """ Update a category """ db_category = get_category_by_user(db, category_id, user_id) if not db_category: return None update_data = category_in.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_category, field, value) db.add(db_category) db.commit() db.refresh(db_category) return db_category def delete_category( db: Session, category_id: int, user_id: int ) -> Optional[Category]: """ Delete a category """ db_category = get_category_by_user(db, category_id, user_id) if not db_category: return None db.delete(db_category) db.commit() return db_category