_aux_functions_templates

_check_template_duplication(*, user_id, name, version, db) async

Ensure that no WorkflowTemplate with the same (user_id, name, version) already exists.

PARAMETER	DESCRIPTION
user_id	TYPE: int
name	TYPE: str
version	TYPE: int
db	TYPE: AsyncSession

RAISES	DESCRIPTION
HTTPException(status_code=HTTP_422_UNPROCESSABLE_CONTENT)	If a duplicate template is found.

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

async def _check_template_duplication(
    *, user_id: int, name: str, version: int, db: AsyncSession
) -> None:
    """
    Ensure that no `WorkflowTemplate` with the same
    (`user_id`, `name`, `version`) already exists.

    Args:
        user_id:
        name:
        version:
        db:

    Raises:
        HTTPException(status_code=HTTP_422_UNPROCESSABLE_CONTENT):
            If a duplicate template is found.


    """
    res = await db.execute(
        select(WorkflowTemplate)
        .where(WorkflowTemplate.user_id == user_id)
        .where(WorkflowTemplate.name == name)
        .where(WorkflowTemplate.version == version)
    )
    duplicate = res.one_or_none()
    if duplicate:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=(
                "The current user already own a workflow template with "
                f"{name=} and {version=}."
            ),
        )

_get_template_full_access(*, user_id, template_id, db) async

Retrieve a WorkflowTemplate and ensure the user is its owner.

PARAMETER	DESCRIPTION
user_id	TYPE: int
template_id	TYPE: int
db	TYPE: AsyncSession

RETURNS	DESCRIPTION
WorkflowTemplate	The WorkflowTemplate object.

RAISES	DESCRIPTION
HTTPException(status_code=404_NOT_FOUND)	If no template exists with the given ID.
HTTPException(status_code=403_FORBIDDEN)	If the user is not the owner.

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

async def _get_template_full_access(
    *, user_id: int, template_id: int, db: AsyncSession
) -> WorkflowTemplate:
    """
    Retrieve a `WorkflowTemplate` and ensure the user is its owner.

    Args:
        user_id:
        template_id:
        db:

    Returns:
        The `WorkflowTemplate` object.

    Raises:
        HTTPException(status_code=404_NOT_FOUND):
            If no template exists with the given ID.
        HTTPException(status_code=403_FORBIDDEN):
            If the user is not the owner.
    """
    template = await _get_template_or_404(template_id=template_id, db=db)
    if template.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=(
                f"You are not the owner of WorkflowTemplate[{template_id}]."
            ),
        )
    return template

_get_template_or_404(*, template_id, db) async

Retrieve a WorkflowTemplate by ID.

PARAMETER	DESCRIPTION
template_id	TYPE: int
db	TYPE: AsyncSession

RETURNS	DESCRIPTION
WorkflowTemplate	The WorkflowTemplate object.

RAISES	DESCRIPTION
HTTPException(status_code=404_NOT_FOUND)	If no template exists with the given ID.

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

async def _get_template_or_404(
    *, template_id: int, db: AsyncSession
) -> WorkflowTemplate:
    """
    Retrieve a `WorkflowTemplate` by ID.

    Args:
        template_id:
        db:

    Returns:
        The `WorkflowTemplate` object.

    Raises:
        HTTPException(status_code=404_NOT_FOUND):
            If no template exists with the given ID.
    """
    template = await db.get(WorkflowTemplate, template_id)
    if template is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"WorkflowTemplate[{template_id}] not found.",
        )
    return template

_get_template_read_access(*, user_id, template_id, db) async

Retrieve a WorkflowTemplate and ensure the user has read access.

Access is granted if the user is the owner of the template or belongs to the template's user group.

PARAMETER	DESCRIPTION
user_id	TYPE: int
template_id	TYPE: int
db	TYPE: AsyncSession

RETURNS	DESCRIPTION
WorkflowTemplate	The WorkflowTemplate object.

RAISES	DESCRIPTION
HTTPException(status_code=404_NOT_FOUND)	If no template exists with the given ID.
HTTPException(status_code=403_FORBIDDEN)	If the user has not read access.

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

async def _get_template_read_access(
    *, user_id: int, template_id: int, db: AsyncSession
) -> WorkflowTemplate:
    """
    Retrieve a `WorkflowTemplate` and ensure the user has read access.

    Access is granted if the user is the owner of the template or belongs to
    the template's user group.

    Args:
        user_id:
        template_id:
        db:

    Returns:
        The `WorkflowTemplate` object.

    Raises:
        HTTPException(status_code=404_NOT_FOUND):
            If no template exists with the given ID.
        HTTPException(status_code=403_FORBIDDEN):
            If the user has not read access.
    """
    template = await _get_template_or_404(template_id=template_id, db=db)
    if template.user_id != user_id:
        link = await db.get(LinkUserGroup, (template.user_group_id, user_id))
        if link is None:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=(
                    "You are not authorized to view "
                    f"WorkflowTemplate[{template_id}]."
                ),
            )
    return template