Cheng Zhou
51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select, update as sa_update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.imp_product import ImpProduct
|
|
from app.schemas.imp import ProductCreate, ProductUpdate
|
|
|
|
|
|
async def create_product(db: AsyncSession, study_id: uuid.UUID, product_in: ProductCreate) -> ImpProduct:
|
|
product = ImpProduct(
|
|
study_id=study_id,
|
|
name=product_in.name,
|
|
form=product_in.form,
|
|
strength=product_in.strength,
|
|
unit=product_in.unit,
|
|
description=product_in.description,
|
|
is_active=product_in.is_active,
|
|
)
|
|
db.add(product)
|
|
await db.commit()
|
|
await db.refresh(product)
|
|
return product
|
|
|
|
|
|
async def get_product(db: AsyncSession, product_id: uuid.UUID) -> ImpProduct | None:
|
|
result = await db.execute(select(ImpProduct).where(ImpProduct.id == product_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_products(db: AsyncSession, study_id: uuid.UUID, is_active: bool | None = None) -> Sequence[ImpProduct]:
|
|
stmt = select(ImpProduct).where(ImpProduct.study_id == study_id)
|
|
if is_active is not None:
|
|
stmt = stmt.where(ImpProduct.is_active == is_active)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_product(db: AsyncSession, product: ImpProduct, product_in: ProductUpdate) -> ImpProduct:
|
|
update_data = product_in.model_dump(exclude_unset=True)
|
|
if update_data:
|
|
await db.execute(
|
|
sa_update(ImpProduct)
|
|
.where(ImpProduct.id == product.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(product)
|
|
return product
|