Cheng Zhou
55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.milestone import Milestone
|
|
from app.schemas.project_milestone import ProjectMilestoneUpdate
|
|
|
|
PROJECT_MILESTONE_TYPE = "SETUP_PROJECT_MILESTONE"
|
|
|
|
|
|
async def list_project_milestones(db: AsyncSession, study_id: uuid.UUID) -> Sequence[Milestone]:
|
|
stmt = (
|
|
select(Milestone)
|
|
.where(
|
|
Milestone.study_id == study_id,
|
|
Milestone.type == PROJECT_MILESTONE_TYPE,
|
|
)
|
|
.order_by(Milestone.planned_date.asc().nulls_last(), Milestone.created_at.asc())
|
|
)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_project_milestone(db: AsyncSession, study_id: uuid.UUID, milestone_id: uuid.UUID) -> Milestone | None:
|
|
result = await db.execute(
|
|
select(Milestone).where(
|
|
Milestone.id == milestone_id,
|
|
Milestone.study_id == study_id,
|
|
Milestone.type == PROJECT_MILESTONE_TYPE,
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_project_milestone(db: AsyncSession, milestone: Milestone, data: ProjectMilestoneUpdate) -> Milestone:
|
|
milestone.adjusted_start_date = data.adjusted_start_date
|
|
milestone.adjusted_end_date = data.adjusted_end_date
|
|
milestone.actual_start_date = data.actual_start_date
|
|
milestone.actual_end_date = data.actual_end_date
|
|
milestone.actual_date = data.actual_end_date
|
|
milestone.notes = data.notes
|
|
|
|
if milestone.actual_end_date:
|
|
milestone.status = "DONE"
|
|
elif milestone.actual_start_date:
|
|
milestone.status = "IN_PROGRESS"
|
|
else:
|
|
milestone.status = "NOT_STARTED"
|
|
|
|
await db.commit()
|
|
await db.refresh(milestone)
|
|
return milestone
|