Cheng Zhou
76 lines
2.3 KiB
Python
76 lines
2.3 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.study import Study
|
|
from app.schemas.study import StudyCreate, StudyUpdate
|
|
|
|
|
|
async def create(db: AsyncSession, study_in: StudyCreate, *, created_by: uuid.UUID | None = None) -> Study:
|
|
study = Study(
|
|
code=study_in.code,
|
|
name=study_in.name,
|
|
sponsor=study_in.sponsor,
|
|
protocol_no=study_in.protocol_no,
|
|
phase=study_in.phase,
|
|
status=study_in.status,
|
|
visit_interval_days=study_in.visit_interval_days,
|
|
visit_total=study_in.visit_total,
|
|
visit_window_start_offset=study_in.visit_window_start_offset,
|
|
visit_window_end_offset=study_in.visit_window_end_offset,
|
|
created_by=created_by,
|
|
)
|
|
db.add(study)
|
|
await db.commit()
|
|
await db.refresh(study)
|
|
return study
|
|
|
|
|
|
async def get(db: AsyncSession, study_id: uuid.UUID) -> Study | None:
|
|
result = await db.execute(select(Study).where(Study.id == study_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_by_code(db: AsyncSession, code: str) -> Study | None:
|
|
result = await db.execute(select(Study).where(Study.code == code))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update(db: AsyncSession, study: Study, study_in: StudyUpdate) -> Study:
|
|
update_data = study_in.model_dump(exclude_unset=True)
|
|
if update_data:
|
|
await db.execute(
|
|
sa_update(Study)
|
|
.where(Study.id == study.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(study)
|
|
return study
|
|
|
|
|
|
async def list_studies(db: AsyncSession, skip: int = 0, limit: int = 100) -> Sequence[Study]:
|
|
result = await db.execute(select(Study).offset(skip).limit(limit))
|
|
return result.scalars().all()
|
|
|
|
|
|
async def list_studies_for_user(
|
|
db: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Sequence[Study]:
|
|
from app.models.study_member import StudyMember # local import to avoid cycles
|
|
|
|
stmt = (
|
|
select(Study)
|
|
.join(StudyMember, StudyMember.study_id == Study.id)
|
|
.where(StudyMember.user_id == user_id, StudyMember.is_active.is_(True))
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|