Cheng Zhou
27 lines
862 B
Python
27 lines
862 B
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.workflow_template import WorkflowTemplate
|
|
|
|
|
|
async def get(db: AsyncSession, template_id: uuid.UUID) -> WorkflowTemplate | None:
|
|
result = await db.execute(select(WorkflowTemplate).where(WorkflowTemplate.id == template_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_active(db: AsyncSession, trial_id: uuid.UUID | None = None) -> Sequence[WorkflowTemplate]:
|
|
stmt = (
|
|
select(WorkflowTemplate)
|
|
.where(WorkflowTemplate.is_active.is_(True))
|
|
.options(selectinload(WorkflowTemplate.nodes))
|
|
)
|
|
if trial_id:
|
|
stmt = stmt.where(WorkflowTemplate.trial_id == trial_id)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|