Step 8:数据管理(Queries + SDV/SDR 进度)

This commit is contained in:
Cheng Zhou
2025-12-16 19:16:50 +08:00
parent 8383fd1dd5
commit 566ebdf749
64 changed files with 646 additions and 1 deletions
+153
View File
@@ -0,0 +1,153 @@
import uuid
from datetime import date
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.deps import get_current_user, get_db_session, require_study_member, require_study_roles
from app.crud import data_query as dq_crud
from app.crud import audit as audit_crud
from app.crud import study as study_crud
from app.schemas.data_query import DataQueryCreate, DataQueryRead, DataQueryUpdate
router = APIRouter()
def _is_overdue(dq: DataQueryRead) -> bool:
return bool(dq.due_date and date.today() > dq.due_date and dq.status != "CLOSED")
async def _ensure_study_exists(db: AsyncSession, study_id: uuid.UUID):
study = await study_crud.get(db, study_id)
if not study:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Study not found")
return study
@router.post(
"/",
response_model=DataQueryRead,
status_code=status.HTTP_201_CREATED,
dependencies=[Depends(require_study_roles(["PM", "CRA"]))],
)
async def create_query(
study_id: uuid.UUID,
query_in: DataQueryCreate,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> DataQueryRead:
await _ensure_study_exists(db, study_id)
try:
dq = await dq_crud.create_query(db, study_id, query_in, created_by=current_user.id)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
await audit_crud.log_action(
db,
study_id=study_id,
entity_type="data_query",
entity_id=dq.id,
action="CREATE_DATA_QUERY",
detail=f"DataQuery {dq.id} created",
operator_id=current_user.id,
operator_role=current_user.role,
)
obj = DataQueryRead.model_validate(dq)
obj.is_overdue = _is_overdue(obj)
return obj
@router.get(
"/",
response_model=list[DataQueryRead],
dependencies=[Depends(require_study_member())],
)
async def list_queries(
study_id: uuid.UUID,
status_filter: str | None = None,
site_id: uuid.UUID | None = None,
subject_id: uuid.UUID | None = None,
assigned_to: uuid.UUID | None = None,
overdue: bool | None = None,
category: str | None = None,
priority: str | None = None,
skip: int = 0,
limit: int = 100,
db: AsyncSession = Depends(get_db_session),
) -> list[DataQueryRead]:
await _ensure_study_exists(db, study_id)
queries = await dq_crud.list_queries(
db,
study_id,
status=status_filter,
site_id=site_id,
subject_id=subject_id,
assigned_to=assigned_to,
overdue=overdue,
category=category,
priority=priority,
skip=skip,
limit=limit,
)
result: list[DataQueryRead] = []
for q in queries:
obj = DataQueryRead.model_validate(q)
obj.is_overdue = _is_overdue(obj)
result.append(obj)
return result
@router.get(
"/{query_id}",
response_model=DataQueryRead,
dependencies=[Depends(require_study_member())],
)
async def get_query(
study_id: uuid.UUID,
query_id: uuid.UUID,
db: AsyncSession = Depends(get_db_session),
) -> DataQueryRead:
await _ensure_study_exists(db, study_id)
dq = await dq_crud.get_query(db, query_id)
if not dq or dq.study_id != study_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="DataQuery not found")
obj = DataQueryRead.model_validate(dq)
obj.is_overdue = _is_overdue(obj)
return obj
@router.patch(
"/{query_id}",
response_model=DataQueryRead,
dependencies=[Depends(require_study_roles(["PM", "CRA"]))],
)
async def update_query(
study_id: uuid.UUID,
query_id: uuid.UUID,
query_in: DataQueryUpdate,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> DataQueryRead:
await _ensure_study_exists(db, study_id)
dq = await dq_crud.get_query(db, query_id)
if not dq or dq.study_id != study_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="DataQuery not found")
old_status = dq.status
updated = await dq_crud.update_query(db, dq, query_in)
detail = None
action = "UPDATE_DATA_QUERY"
if query_in.status and query_in.status != old_status:
detail = f"DataQuery {query_id} status {old_status} -> {query_in.status}"
action = "DATA_QUERY_STATUS_CHANGE"
await audit_crud.log_action(
db,
study_id=study_id,
entity_type="data_query",
entity_id=query_id,
action=action,
detail=detail or "DataQuery updated",
operator_id=current_user.id,
operator_role=current_user.role,
)
obj = DataQueryRead.model_validate(updated)
obj.is_overdue = _is_overdue(obj)
return obj
+3 -1
View File
@@ -1,6 +1,6 @@
from fastapi import APIRouter
from app.api.v1 import auth, users, studies, sites, members, comments, attachments, audit_logs, milestones, tasks, dashboard, subjects, visits, aes, issues
from app.api.v1 import auth, users, studies, sites, members, comments, attachments, audit_logs, milestones, tasks, dashboard, subjects, visits, aes, issues, data_queries, verifications
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
@@ -18,3 +18,5 @@ api_router.include_router(subjects.router, prefix="/studies/{study_id}/subjects"
api_router.include_router(visits.router, prefix="/studies/{study_id}/subjects/{subject_id}/visits", tags=["visits"])
api_router.include_router(aes.router, prefix="/studies/{study_id}/aes", tags=["aes"])
api_router.include_router(issues.router, prefix="/studies/{study_id}/issues", tags=["issues"])
api_router.include_router(data_queries.router, prefix="/studies/{study_id}/data-queries", tags=["data-queries"])
api_router.include_router(verifications.router, prefix="/studies/{study_id}/verifications", tags=["verifications"])
+115
View File
@@ -0,0 +1,115 @@
import uuid
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.deps import get_current_user, get_db_session, require_study_member, require_study_roles
from app.crud import audit as audit_crud
from app.crud import verification as verification_crud
from app.crud import study as study_crud
from app.schemas.verification import VerificationCreate, VerificationRead, VerificationUpdate
router = APIRouter()
async def _ensure_study_exists(db: AsyncSession, study_id: uuid.UUID):
study = await study_crud.get(db, study_id)
if not study:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Study not found")
return study
@router.put(
"/",
response_model=VerificationRead,
status_code=status.HTTP_200_OK,
dependencies=[Depends(require_study_roles(["PM", "CRA"]))],
)
async def upsert_verification(
study_id: uuid.UUID,
payload: VerificationCreate,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> VerificationRead:
await _ensure_study_exists(db, study_id)
try:
vp = await verification_crud.upsert_progress(db, study_id, payload)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
await audit_crud.log_action(
db,
study_id=study_id,
entity_type="verification",
entity_id=vp.id,
action="UPSERT_VERIFICATION",
detail=f"{payload.level} progress subject {payload.subject_id} upserted",
operator_id=current_user.id,
operator_role=current_user.role,
)
return VerificationRead.model_validate(vp)
@router.get(
"/",
response_model=list[VerificationRead],
dependencies=[Depends(require_study_member())],
)
async def list_verifications(
study_id: uuid.UUID,
site_id: uuid.UUID | None = None,
subject_id: uuid.UUID | None = None,
level: str | None = None,
db: AsyncSession = Depends(get_db_session),
) -> list[VerificationRead]:
await _ensure_study_exists(db, study_id)
items = await verification_crud.list_progress(db, study_id, site_id=site_id, subject_id=subject_id, level=level)
return [VerificationRead.model_validate(i) for i in items]
@router.get(
"/subjects/{subject_id}/{level}",
response_model=VerificationRead,
dependencies=[Depends(require_study_member())],
)
async def get_verification(
study_id: uuid.UUID,
subject_id: uuid.UUID,
level: str,
db: AsyncSession = Depends(get_db_session),
) -> VerificationRead:
await _ensure_study_exists(db, study_id)
vp = await verification_crud.get_progress(db, study_id, subject_id, level)
if not vp:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Verification not found")
return VerificationRead.model_validate(vp)
@router.patch(
"/{verification_id}",
response_model=VerificationRead,
dependencies=[Depends(require_study_roles(["PM", "CRA"]))],
)
async def update_verification(
study_id: uuid.UUID,
verification_id: uuid.UUID,
payload: VerificationUpdate,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> VerificationRead:
await _ensure_study_exists(db, study_id)
vp = await verification_crud.get_by_id(db, verification_id)
if not vp or vp.study_id != study_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Verification not found")
updated = await verification_crud.update_progress(db, vp, payload)
await audit_crud.log_action(
db,
study_id=study_id,
entity_type="verification",
entity_id=verification_id,
action="UPSERT_VERIFICATION",
detail=f"{updated.level} progress subject {updated.subject_id} updated",
operator_id=current_user.id,
operator_role=current_user.role,
)
return VerificationRead.model_validate(updated)
+117
View File
@@ -0,0 +1,117 @@
import uuid
from datetime import date, datetime, timezone
from typing import Sequence
from sqlalchemy import select, update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.data_query import DataQuery
from app.models.site import Site
from app.models.subject import Subject
from app.schemas.data_query import DataQueryCreate, DataQueryUpdate
async def _validate_site_subject(db: AsyncSession, study_id: uuid.UUID, site_id: uuid.UUID | None, subject_id: uuid.UUID | None):
if site_id:
result = await db.execute(select(Site).where(Site.id == site_id))
site = result.scalar_one_or_none()
if not site or site.study_id != study_id:
raise ValueError("Site not found in study")
subject_site = None
if subject_id:
result = await db.execute(select(Subject).where(Subject.id == subject_id))
subj = result.scalar_one_or_none()
if not subj or subj.study_id != study_id:
raise ValueError("Subject not found in study")
subject_site = subj.site_id
if site_id and subject_site and site_id != subject_site:
raise ValueError("Site and subject mismatch")
async def create_query(
db: AsyncSession,
study_id: uuid.UUID,
query_in: DataQueryCreate,
*,
created_by: uuid.UUID,
) -> DataQuery:
await _validate_site_subject(db, study_id, query_in.site_id, query_in.subject_id)
dq = DataQuery(
study_id=study_id,
site_id=query_in.site_id,
subject_id=query_in.subject_id,
visit_id=query_in.visit_id,
title=query_in.title,
description=query_in.description,
category=query_in.category,
priority=query_in.priority,
assigned_to=query_in.assigned_to,
due_date=query_in.due_date,
status="OPEN",
resolution=None,
closed_at=None,
created_by=created_by,
)
db.add(dq)
await db.commit()
await db.refresh(dq)
return dq
async def get_query(db: AsyncSession, query_id: uuid.UUID) -> DataQuery | None:
result = await db.execute(select(DataQuery).where(DataQuery.id == query_id))
return result.scalar_one_or_none()
async def list_queries(
db: AsyncSession,
study_id: uuid.UUID,
*,
status: str | None = None,
site_id: uuid.UUID | None = None,
subject_id: uuid.UUID | None = None,
assigned_to: uuid.UUID | None = None,
overdue: bool | None = None,
category: str | None = None,
priority: str | None = None,
skip: int = 0,
limit: int = 100,
) -> Sequence[DataQuery]:
stmt = select(DataQuery).where(DataQuery.study_id == study_id)
if status:
stmt = stmt.where(DataQuery.status == status)
if site_id:
stmt = stmt.where(DataQuery.site_id == site_id)
if subject_id:
stmt = stmt.where(DataQuery.subject_id == subject_id)
if assigned_to:
stmt = stmt.where(DataQuery.assigned_to == assigned_to)
if category:
stmt = stmt.where(DataQuery.category == category)
if priority:
stmt = stmt.where(DataQuery.priority == priority)
if overdue is True:
stmt = stmt.where(DataQuery.due_date < date.today(), DataQuery.status != "CLOSED")
if overdue is False:
stmt = stmt.where((DataQuery.due_date >= date.today()) | (DataQuery.due_date.is_(None)) | (DataQuery.status == "CLOSED"))
stmt = stmt.offset(skip).limit(limit)
result = await db.execute(stmt)
return result.scalars().all()
async def update_query(db: AsyncSession, dq: DataQuery, dq_in: DataQueryUpdate) -> DataQuery:
update_data = dq_in.model_dump(exclude_unset=True)
if "status" in update_data:
if update_data["status"] == "CLOSED":
update_data["closed_at"] = datetime.now(timezone.utc)
else:
update_data["closed_at"] = None
if update_data:
await db.execute(
sa_update(DataQuery)
.where(DataQuery.id == dq.id)
.values(**update_data)
)
await db.commit()
await db.refresh(dq)
return dq
+108
View File
@@ -0,0 +1,108 @@
import uuid
from datetime import date
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.verification import VerificationProgress
from app.schemas.verification import VerificationCreate, VerificationUpdate
async def _validate_site_subject(db: AsyncSession, study_id: uuid.UUID, site_id: uuid.UUID, subject_id: uuid.UUID):
result = await db.execute(select(Subject).where(Subject.id == subject_id))
subj = result.scalar_one_or_none()
if not subj or subj.study_id != study_id:
raise ValueError("Subject not found in study")
if subj.site_id != site_id:
raise ValueError("Site does not match subject")
async def upsert_progress(
db: AsyncSession,
study_id: uuid.UUID,
data: VerificationCreate,
) -> VerificationProgress:
await _validate_site_subject(db, study_id, data.site_id, data.subject_id)
result = await db.execute(
select(VerificationProgress).where(
VerificationProgress.study_id == study_id,
VerificationProgress.subject_id == data.subject_id,
VerificationProgress.level == data.level,
)
)
existing = result.scalar_one_or_none()
if existing:
update_data = data.model_dump()
await db.execute(
sa_update(VerificationProgress)
.where(VerificationProgress.id == existing.id)
.values(**update_data)
)
await db.commit()
await db.refresh(existing)
return existing
vp = VerificationProgress(
study_id=study_id,
site_id=data.site_id,
subject_id=data.subject_id,
level=data.level,
percent=data.percent,
last_verified_at=data.last_verified_at,
verifier_id=data.verifier_id,
notes=data.notes,
)
db.add(vp)
await db.commit()
await db.refresh(vp)
return vp
async def get_progress(db: AsyncSession, study_id: uuid.UUID, subject_id: uuid.UUID, level: str) -> VerificationProgress | None:
result = await db.execute(
select(VerificationProgress).where(
VerificationProgress.study_id == study_id,
VerificationProgress.subject_id == subject_id,
VerificationProgress.level == level,
)
)
return result.scalar_one_or_none()
async def get_by_id(db: AsyncSession, verification_id: uuid.UUID) -> VerificationProgress | None:
result = await db.execute(select(VerificationProgress).where(VerificationProgress.id == verification_id))
return result.scalar_one_or_none()
async def list_progress(
db: AsyncSession,
study_id: uuid.UUID,
site_id: uuid.UUID | None = None,
subject_id: uuid.UUID | None = None,
level: str | None = None,
) -> Sequence[VerificationProgress]:
stmt = select(VerificationProgress).where(VerificationProgress.study_id == study_id)
if site_id:
stmt = stmt.where(VerificationProgress.site_id == site_id)
if subject_id:
stmt = stmt.where(VerificationProgress.subject_id == subject_id)
if level:
stmt = stmt.where(VerificationProgress.level == level)
result = await db.execute(stmt)
return result.scalars().all()
async def update_progress(db: AsyncSession, vp: VerificationProgress, data: VerificationUpdate) -> VerificationProgress:
update_data = data.model_dump(exclude_unset=True)
if update_data:
await db.execute(
sa_update(VerificationProgress)
.where(VerificationProgress.id == vp.id)
.values(**update_data)
)
await db.commit()
await db.refresh(vp)
return vp
+2
View File
@@ -14,3 +14,5 @@ from app.models.subject import Subject # noqa: F401
from app.models.visit import Visit # noqa: F401
from app.models.ae import AdverseEvent # noqa: F401
from app.models.issue import Issue # noqa: F401
from app.models.data_query import DataQuery # noqa: F401
from app.models.verification import VerificationProgress # noqa: F401
+32
View File
@@ -0,0 +1,32 @@
import uuid
from datetime import datetime, date
from sqlalchemy import Date, DateTime, ForeignKey, String, Text, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base_class import Base
class DataQuery(Base):
__tablename__ = "data_queries"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
study_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("studies.id"), index=True, nullable=False)
site_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("sites.id"), index=True, nullable=True)
subject_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("subjects.id"), index=True, nullable=True)
visit_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), nullable=True)
title: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
category: Mapped[str] = mapped_column(String(50), nullable=False)
priority: Mapped[str] = mapped_column(String(20), nullable=False)
assigned_to: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
due_date: Mapped[date | None] = mapped_column(Date, nullable=True)
status: Mapped[str] = mapped_column(String(20), nullable=False, default="OPEN")
resolution: Mapped[str | None] = mapped_column(Text, nullable=True)
closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
created_by: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
)
+27
View File
@@ -0,0 +1,27 @@
import uuid
from datetime import datetime, date
from sqlalchemy import Date, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base_class import Base
class VerificationProgress(Base):
__tablename__ = "verification_progress"
__table_args__ = (UniqueConstraint("study_id", "subject_id", "level", name="uq_verification_subject_level"),)
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
study_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("studies.id"), index=True, nullable=False)
site_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("sites.id"), index=True, nullable=False)
subject_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("subjects.id"), index=True, nullable=False)
level: Mapped[str] = mapped_column(String(10), nullable=False) # SDR / SDV
percent: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
last_verified_at: Mapped[date | None] = mapped_column(Date, nullable=True)
verifier_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
notes: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
)
+51
View File
@@ -0,0 +1,51 @@
import uuid
from datetime import date, datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict
class DataQueryCreate(BaseModel):
title: str
description: Optional[str] = None
category: str
priority: str
site_id: Optional[uuid.UUID] = None
subject_id: Optional[uuid.UUID] = None
visit_id: Optional[uuid.UUID] = None
assigned_to: Optional[uuid.UUID] = None
due_date: Optional[date] = None
class DataQueryUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
category: Optional[str] = None
priority: Optional[str] = None
assigned_to: Optional[uuid.UUID] = None
due_date: Optional[date] = None
status: Optional[str] = None
resolution: Optional[str] = None
class DataQueryRead(BaseModel):
id: uuid.UUID
study_id: uuid.UUID
site_id: Optional[uuid.UUID]
subject_id: Optional[uuid.UUID]
visit_id: Optional[uuid.UUID]
title: str
description: Optional[str]
category: str
priority: str
assigned_to: Optional[uuid.UUID]
due_date: Optional[date]
status: str
resolution: Optional[str]
closed_at: Optional[datetime]
created_by: uuid.UUID
created_at: datetime
updated_at: datetime
is_overdue: bool | None = None
model_config = ConfigDict(from_attributes=True)
+38
View File
@@ -0,0 +1,38 @@
import uuid
from datetime import date, datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class VerificationCreate(BaseModel):
site_id: uuid.UUID
subject_id: uuid.UUID
level: str
percent: int = Field(ge=0, le=100)
last_verified_at: Optional[date] = None
verifier_id: Optional[uuid.UUID] = None
notes: Optional[str] = None
class VerificationUpdate(BaseModel):
percent: Optional[int] = Field(default=None, ge=0, le=100)
last_verified_at: Optional[date] = None
verifier_id: Optional[uuid.UUID] = None
notes: Optional[str] = None
class VerificationRead(BaseModel):
id: uuid.UUID
study_id: uuid.UUID
site_id: uuid.UUID
subject_id: uuid.UUID
level: str
percent: int
last_verified_at: Optional[date]
verifier_id: Optional[uuid.UUID]
notes: Optional[str]
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)