Cheng Zhou
71 lines
1.9 KiB
Python
71 lines
1.9 KiB
Python
import uuid
|
|
from datetime import date
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.imp_transaction import ImpTransaction
|
|
|
|
|
|
async def create_tx(
|
|
db: AsyncSession,
|
|
*,
|
|
study_id: uuid.UUID,
|
|
site_id: uuid.UUID,
|
|
batch_id: uuid.UUID,
|
|
subject_id: uuid.UUID | None,
|
|
tx_type: str,
|
|
quantity: int,
|
|
tx_date: date,
|
|
reference: str | None,
|
|
notes: str | None,
|
|
created_by: uuid.UUID,
|
|
) -> ImpTransaction:
|
|
tx = ImpTransaction(
|
|
study_id=study_id,
|
|
site_id=site_id,
|
|
batch_id=batch_id,
|
|
subject_id=subject_id,
|
|
tx_type=tx_type,
|
|
quantity=quantity,
|
|
tx_date=tx_date,
|
|
reference=reference,
|
|
notes=notes,
|
|
created_by=created_by,
|
|
)
|
|
db.add(tx)
|
|
await db.flush()
|
|
return tx
|
|
|
|
|
|
async def list_txs(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
*,
|
|
site_id: uuid.UUID | None = None,
|
|
batch_id: uuid.UUID | None = None,
|
|
subject_id: uuid.UUID | None = None,
|
|
tx_type: str | None = None,
|
|
date_from: date | None = None,
|
|
date_to: date | None = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Sequence[ImpTransaction]:
|
|
stmt = select(ImpTransaction).where(ImpTransaction.study_id == study_id)
|
|
if site_id:
|
|
stmt = stmt.where(ImpTransaction.site_id == site_id)
|
|
if batch_id:
|
|
stmt = stmt.where(ImpTransaction.batch_id == batch_id)
|
|
if subject_id:
|
|
stmt = stmt.where(ImpTransaction.subject_id == subject_id)
|
|
if tx_type:
|
|
stmt = stmt.where(ImpTransaction.tx_type == tx_type)
|
|
if date_from:
|
|
stmt = stmt.where(ImpTransaction.tx_date >= date_from)
|
|
if date_to:
|
|
stmt = stmt.where(ImpTransaction.tx_date <= date_to)
|
|
stmt = stmt.offset(skip).limit(limit)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|