61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.category import Category
|
|
from app.schemas.category import CategoryCreate, CategoryUpdate
|
|
|
|
|
|
async def get_categories(db: AsyncSession) -> List[Category]:
|
|
"""Get all categories."""
|
|
result = await db.execute(select(Category))
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_category(db: AsyncSession, category_id: int) -> Optional[Category]:
|
|
"""Get a category by ID."""
|
|
return await db.get(Category, category_id)
|
|
|
|
|
|
async def get_category_with_products(
|
|
db: AsyncSession, category_id: int
|
|
) -> Optional[Category]:
|
|
"""Get a category with related products."""
|
|
query = (
|
|
select(Category)
|
|
.where(Category.id == category_id)
|
|
.options(selectinload(Category.products))
|
|
)
|
|
result = await db.execute(query)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def create_category(db: AsyncSession, category: CategoryCreate) -> Category:
|
|
"""Create a new category."""
|
|
db_category = Category(**category.model_dump())
|
|
db.add(db_category)
|
|
await db.commit()
|
|
await db.refresh(db_category)
|
|
return db_category
|
|
|
|
|
|
async def update_category(
|
|
db: AsyncSession, db_category: Category, category_update: CategoryUpdate
|
|
) -> Category:
|
|
"""Update a category."""
|
|
update_data = category_update.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_category, field, value)
|
|
|
|
await db.commit()
|
|
await db.refresh(db_category)
|
|
return db_category
|
|
|
|
|
|
async def delete_category(db: AsyncSession, db_category: Category) -> None:
|
|
"""Delete a category."""
|
|
await db.delete(db_category)
|
|
await db.commit()
|