_aux_functions_history

_verify_workflow_and_dataset_access(*, project_id, workflow_id, dataset_id, user_id, db) async

Verify user access to a dataset/workflow pair.

Parameters:

Name	Type	Description	Default
dataset_id	int		required
workflow_task_id			required
user_id	int		required
db	AsyncSession		required

Source code in fractal_server/app/routes/api/v2/_aux_functions_history.py

async def _verify_workflow_and_dataset_access(
    *,
    project_id: int,
    workflow_id: int,
    dataset_id: int,
    user_id: int,
    db: AsyncSession,
) -> dict[Literal["dataset", "workflow"], DatasetV2 | WorkflowV2]:
    """
    Verify user access to a dataset/workflow pair.

    Args:
        dataset_id:
        workflow_task_id:
        user_id:
        db:
    """
    await _get_project_check_owner(
        project_id=project_id,
        user_id=user_id,
        db=db,
    )
    workflow = await _get_workflow_or_404(
        workflow_id=workflow_id,
        db=db,
    )
    if workflow.project_id != project_id:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="Workflow does not belong to expected project.",
        )
    dataset = await _get_dataset_or_404(
        dataset_id=dataset_id,
        db=db,
    )
    if dataset.project_id != project_id:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="Dataset does not belong to expected project.",
        )

    return dict(dataset=dataset, workflow=workflow)

get_history_run_or_404(*, history_run_id, db) async

Get an existing HistoryRun or raise a 404.

Parameters:

Name	Type	Description	Default
history_run_id	int		required
db	AsyncSession		required

Source code in fractal_server/app/routes/api/v2/_aux_functions_history.py

async def get_history_run_or_404(
    *, history_run_id: int, db: AsyncSession
) -> HistoryRun:
    """
    Get an existing HistoryRun  or raise a 404.

    Arguments:
        history_run_id:
        db:
    """
    history_run = await db.get(HistoryRun, history_run_id)
    if history_run is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"HistoryRun {history_run_id} not found",
        )
    return history_run

get_history_unit_or_404(*, history_unit_id, db) async

Get an existing HistoryUnit or raise a 404.

Parameters:

Name	Type	Description	Default
history_unit_id	int	The HistoryUnit id	required
db	AsyncSession	An asynchronous db session	required

Source code in fractal_server/app/routes/api/v2/_aux_functions_history.py

async def get_history_unit_or_404(
    *, history_unit_id: int, db: AsyncSession
) -> HistoryUnit:
    """
    Get an existing HistoryUnit  or raise a 404.

    Arguments:
        history_unit_id: The `HistoryUnit` id
        db: An asynchronous db session
    """
    history_unit = await db.get(HistoryUnit, history_unit_id)
    if history_unit is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"HistoryUnit {history_unit_id} not found",
        )
    return history_unit

get_wftask_check_owner(*, project_id, dataset_id, workflowtask_id, user_id, db) async

Verify user access for the history of this dataset and workflowtask.

Parameters:

Name	Type	Description	Default
project_id	int		required
dataset_id	int		required
workflow_task_id			required
user_id	int		required
db	AsyncSession		required

Source code in fractal_server/app/routes/api/v2/_aux_functions_history.py

async def get_wftask_check_owner(
    *,
    project_id: int,
    dataset_id: int,
    workflowtask_id: int,
    user_id: int,
    db: AsyncSession,
) -> WorkflowTaskV2:
    """
    Verify user access for the history of this dataset and workflowtask.

    Args:
        project_id:
        dataset_id:
        workflow_task_id:
        user_id:
        db:
    """
    wftask = await _get_workflowtask_or_404(
        workflowtask_id=workflowtask_id,
        db=db,
    )
    await _verify_workflow_and_dataset_access(
        project_id=project_id,
        dataset_id=dataset_id,
        workflow_id=wftask.workflow_id,
        user_id=user_id,
        db=db,
    )
    return wftask