Cheng Zhou
66 lines
1.8 KiB
Python
66 lines
1.8 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.subject import Subject
|
|
from app.models.visit import Visit
|
|
from app.schemas.visit import VisitCreate, VisitUpdate
|
|
|
|
|
|
async def create_visit(
|
|
db: AsyncSession,
|
|
*,
|
|
study_id: uuid.UUID,
|
|
visit_in: VisitCreate | None,
|
|
subject: Subject,
|
|
visit_code: str,
|
|
visit_name: str,
|
|
planned_date,
|
|
) -> Visit:
|
|
visit = Visit(
|
|
study_id=study_id,
|
|
subject_id=subject.id,
|
|
visit_code=visit_code,
|
|
visit_name=visit_name,
|
|
planned_date=planned_date,
|
|
actual_date=None,
|
|
status="PLANNED",
|
|
window_start=visit_in.window_start if visit_in else None,
|
|
window_end=visit_in.window_end if visit_in else None,
|
|
notes=None,
|
|
)
|
|
db.add(visit)
|
|
await db.commit()
|
|
await db.refresh(visit)
|
|
return visit
|
|
|
|
|
|
async def list_visits(db: AsyncSession, subject_id: uuid.UUID) -> Sequence[Visit]:
|
|
result = await db.execute(select(Visit).where(Visit.subject_id == subject_id).order_by(Visit.planned_date))
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_visit(db: AsyncSession, visit_id: uuid.UUID) -> Visit | None:
|
|
result = await db.execute(select(Visit).where(Visit.id == visit_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_visit(db: AsyncSession, visit: Visit, visit_in: VisitUpdate) -> Visit:
|
|
update_data = visit_in.model_dump(exclude_unset=True)
|
|
if update_data:
|
|
await db.execute(
|
|
sa_update(Visit)
|
|
.where(Visit.id == visit.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(visit)
|
|
return visit
|
|
|
|
|
|
async def delete_visit(db: AsyncSession, visit: Visit) -> None:
|
|
await db.delete(visit)
|
|
await db.commit()
|