43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from core.models import Product
|
|
from sqlalchemy.engine import Result
|
|
|
|
from .schemas import ProductCreate, ProductUpdate, ProductUpdatePartial
|
|
|
|
|
|
async def get_all_products(session: AsyncSession) -> list[Product]:
|
|
stmt = select(Product).order_by(Product.id)
|
|
result: Result = await session.execute(stmt)
|
|
products = result.scalars().all()
|
|
return list(products)
|
|
|
|
|
|
async def get_products_by_id(session: AsyncSession, product_id: int) -> Product | None:
|
|
return await session.get(Product, product_id)
|
|
|
|
|
|
async def create_product(session: AsyncSession, product_in: ProductCreate) -> Product:
|
|
new_product = Product(**product_in.model_dump())
|
|
session.add(new_product)
|
|
await session.commit()
|
|
await session.refresh(new_product)
|
|
return new_product
|
|
|
|
|
|
async def update_product(
|
|
session: AsyncSession,
|
|
product: Product,
|
|
product_update: ProductUpdate | ProductUpdatePartial,
|
|
partial: bool = False,
|
|
) -> Product:
|
|
for name, value in product_update.model_dump(exclude_unset=partial).items():
|
|
setattr(product, name, value)
|
|
await session.commit()
|
|
return product
|
|
|
|
|
|
async def delete_product(session: AsyncSession, product: Product) -> None:
|
|
await session.delete(product)
|
|
await session.commit()
|