Cheng Zhou
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.subject import Subject
|
|
from app.models.subject_history import SubjectHistory
|
|
from app.schemas.subject_history import SubjectHistoryCreate, SubjectHistoryUpdate
|
|
|
|
|
|
async def _ensure_subject(db: AsyncSession, study_id: uuid.UUID, subject_id: uuid.UUID) -> None:
|
|
result = await db.execute(select(Subject).where(Subject.id == subject_id))
|
|
subject = result.scalar_one_or_none()
|
|
if not subject or subject.study_id != study_id:
|
|
raise ValueError("参与者不属于当前项目")
|
|
|
|
|
|
async def create_history(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
history_in: SubjectHistoryCreate,
|
|
created_by: uuid.UUID | None,
|
|
) -> SubjectHistory:
|
|
await _ensure_subject(db, study_id, history_in.subject_id)
|
|
history = SubjectHistory(
|
|
study_id=study_id,
|
|
subject_id=history_in.subject_id,
|
|
record_date=history_in.record_date,
|
|
content=history_in.content,
|
|
created_by=created_by,
|
|
)
|
|
db.add(history)
|
|
await db.commit()
|
|
await db.refresh(history)
|
|
return history
|
|
|
|
|
|
async def get_history(db: AsyncSession, history_id: uuid.UUID) -> SubjectHistory | None:
|
|
result = await db.execute(select(SubjectHistory).where(SubjectHistory.id == history_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_histories(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
subject_id: uuid.UUID,
|
|
skip: int = 0,
|
|
limit: int = 200,
|
|
) -> Sequence[SubjectHistory]:
|
|
stmt = (
|
|
select(SubjectHistory)
|
|
.where(SubjectHistory.study_id == study_id, SubjectHistory.subject_id == subject_id)
|
|
.order_by(SubjectHistory.record_date.desc().nullslast(), SubjectHistory.created_at.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_history(
|
|
db: AsyncSession, history: SubjectHistory, history_in: SubjectHistoryUpdate
|
|
) -> SubjectHistory:
|
|
update_data = history_in.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(history, key, value)
|
|
await db.commit()
|
|
await db.refresh(history)
|
|
return history
|
|
|
|
|
|
async def delete_history(db: AsyncSession, history: SubjectHistory) -> None:
|
|
await db.delete(history)
|
|
await db.commit()
|