Cheng Zhou
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.finance_contract import FinanceContract
|
|
from app.schemas.finance_contract import FinanceContractCreate, FinanceContractUpdate
|
|
|
|
|
|
async def create_contract(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
contract_in: FinanceContractCreate,
|
|
created_by: uuid.UUID | None,
|
|
) -> FinanceContract:
|
|
contract = FinanceContract(
|
|
study_id=study_id,
|
|
site_name=contract_in.site_name,
|
|
contract_no=contract_in.contract_no,
|
|
signed_date=contract_in.signed_date,
|
|
amount=contract_in.amount,
|
|
currency=contract_in.currency,
|
|
remark=contract_in.remark,
|
|
created_by=created_by,
|
|
)
|
|
db.add(contract)
|
|
await db.commit()
|
|
await db.refresh(contract)
|
|
return contract
|
|
|
|
|
|
async def get_contract(db: AsyncSession, contract_id: uuid.UUID) -> FinanceContract | None:
|
|
result = await db.execute(select(FinanceContract).where(FinanceContract.id == contract_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_contracts(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
site_name: str | None = None,
|
|
site_names: set[str] | None = None,
|
|
contract_no: str | None = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Sequence[FinanceContract]:
|
|
stmt = (
|
|
select(FinanceContract)
|
|
.where(FinanceContract.study_id == study_id)
|
|
)
|
|
if site_names is not None:
|
|
if not site_names:
|
|
return []
|
|
stmt = stmt.where(FinanceContract.site_name.in_(site_names))
|
|
if site_name:
|
|
stmt = stmt.where(FinanceContract.site_name.ilike(f"%{site_name}%"))
|
|
if contract_no:
|
|
stmt = stmt.where(FinanceContract.contract_no.ilike(f"%{contract_no}%"))
|
|
stmt = stmt.order_by(FinanceContract.created_at.desc()).offset(skip).limit(limit)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_contract(
|
|
db: AsyncSession, contract: FinanceContract, contract_in: FinanceContractUpdate
|
|
) -> FinanceContract:
|
|
update_data = contract_in.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(contract, key, value)
|
|
await db.commit()
|
|
await db.refresh(contract)
|
|
return contract
|
|
|
|
|
|
async def delete_contract(db: AsyncSession, contract: FinanceContract) -> None:
|
|
await db.delete(contract)
|
|
await db.commit()
|