Skip to content

Repositories

Workflow Runs Repository

create_run_from_workflow(db, workflow)

Create a WorkflowRun model from a Workflow specification.

Source code in germinate_ai/data/repositories/workflow_runs_repository.py 
20
21
22
23
24
25
26
27
28
def create_run_from_workflow(db: Session, workflow: Workflow) -> WorkflowRun:
    """Create a WorkflowRun model from a Workflow specification."""
    workflow_run = WorkflowRun(
        workflow_name=workflow.name,
        workflow_version=workflow.version,
        workflow_id=workflow.workflow_id,
    )
    workflow_run.state = WorkflowRunStateEnum.created
    return workflow_run

get_workflow_run(db, uuid, join_states=True)

Get WorkflowRun from the DB.

Source code in germinate_ai/data/repositories/workflow_runs_repository.py 
12
13
14
15
16
17
def get_workflow_run(db: Session, uuid: UUID, join_states=True) -> WorkflowRun:
    """Get WorkflowRun from the DB."""
    stmt = select(WorkflowRun).where(WorkflowRun.id == uuid)
    if join_states:
        stmt = stmt.join(WorkflowRun.state_instances)
    return db.scalars(stmt).first()

States Repository

get_state(db, uuid, join_tasks=True)

Get State from DB.

Source code in germinate_ai/data/repositories/states_repository.py 
10
11
12
13
14
15
def get_state(db: Session, uuid: UUID, join_tasks=True) -> StateInstance:
    """Get State from DB."""
    stmt = select(StateInstance).where(StateInstance.id == uuid)
    if join_tasks:
        stmt = stmt.join(StateInstance.task_instances)
    return db.scalars(stmt).first()

Tasks Repository

get_task_instance_from_assignment(db, assignment)

Get Task instance from DB from the given assignment.

Source code in germinate_ai/data/repositories/tasks_repository.py 
 9
10
11
12
13
14
15
16
17
18
19
def get_task_instance_from_assignment(
    db: Session, assignment: TaskAssignment
) -> TaskInstance:
    """Get Task instance from DB from the given assignment."""
    stmt = (
        select(TaskInstance)
        .where(TaskInstance.state_instance_id == assignment.state_instance_id)
        .where(TaskInstance.name == assignment.name)
    )
    task = db.scalars(stmt).first()
    return task