Cheng Zhou
71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.precaution import Precaution
|
|
from app.schemas.precaution import PrecautionCreate, PrecautionUpdate
|
|
|
|
|
|
async def create_precaution(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
precaution_in: PrecautionCreate,
|
|
created_by: uuid.UUID | None,
|
|
) -> Precaution:
|
|
precaution = Precaution(
|
|
study_id=study_id,
|
|
site_name=precaution_in.site_name,
|
|
title=precaution_in.title,
|
|
content=precaution_in.content,
|
|
level=precaution_in.level,
|
|
created_by=created_by,
|
|
)
|
|
db.add(precaution)
|
|
await db.commit()
|
|
await db.refresh(precaution)
|
|
return precaution
|
|
|
|
|
|
async def get_precaution(db: AsyncSession, precaution_id: uuid.UUID) -> Precaution | None:
|
|
result = await db.execute(select(Precaution).where(Precaution.id == precaution_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_precautions(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
site_name: str | None = None,
|
|
keyword: str | None = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Sequence[Precaution]:
|
|
stmt = (
|
|
select(Precaution)
|
|
.where(Precaution.study_id == study_id)
|
|
)
|
|
if site_name:
|
|
stmt = stmt.where(Precaution.site_name.ilike(f"%{site_name}%"))
|
|
if keyword:
|
|
stmt = stmt.where(Precaution.title.ilike(f"%{keyword}%"))
|
|
stmt = stmt.order_by(Precaution.updated_at.desc()).offset(skip).limit(limit)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_precaution(
|
|
db: AsyncSession, precaution: Precaution, precaution_in: PrecautionUpdate
|
|
) -> Precaution:
|
|
update_data = precaution_in.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(precaution, key, value)
|
|
await db.commit()
|
|
await db.refresh(precaution)
|
|
return precaution
|
|
|
|
|
|
async def delete_precaution(db: AsyncSession, precaution: Precaution) -> None:
|
|
await db.delete(precaution)
|
|
await db.commit()
|