Skip to content

_aux_functions

Auxiliary functions to get object from the database or perform simple checks

_check_project_exists(*, project_name, user_id, db) async

Check that no other project with this name exists for this user.

Parameters:

Name Type Description Default
project_name str

Project name

required
user_id int

User ID

required
db AsyncSession
required

Raises:

Type Description
HTTPException(status_code=422_UNPROCESSABLE_ENTITY)

If such a project already exists

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
async def _check_project_exists(
    *,
    project_name: str,
    user_id: int,
    db: AsyncSession,
) -> None:
    """
    Check that no other project with this name exists for this user.

    Args:
        project_name: Project name
        user_id: User ID
        db:

    Raises:
        HTTPException(status_code=422_UNPROCESSABLE_ENTITY):
            If such a project already exists
    """
    stm = (
        select(Project)
        .join(LinkUserProject)
        .where(Project.name == project_name)
        .where(LinkUserProject.user_id == user_id)
    )
    res = await db.execute(stm)
    if res.scalars().all():
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Project name ({project_name}) already in use",
        )

_check_workflow_exists(*, name, project_id, db) async

Check that no other workflow of this project has the same name.

Parameters:

Name Type Description Default
name str

Workflow name

required
project_id int

Project ID

required
db AsyncSession
required

Raises:

Type Description
HTTPException(status_code=422_UNPROCESSABLE_ENTITY)

If such a workflow already exists

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
async def _check_workflow_exists(
    *,
    name: str,
    project_id: int,
    db: AsyncSession,
) -> None:
    """
    Check that no other workflow of this project has the same name.

    Args:
        name: Workflow name
        project_id: Project ID
        db:

    Raises:
        HTTPException(status_code=422_UNPROCESSABLE_ENTITY):
            If such a workflow already exists
    """
    stm = (
        select(Workflow)
        .where(Workflow.name == name)
        .where(Workflow.project_id == project_id)
    )
    res = await db.execute(stm)
    if res.scalars().all():
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Workflow with {name=} and {project_id=} already exists.",
        )

_get_dataset_check_owner(*, project_id, dataset_id, user_id, db) async

Get a dataset and a project, after access control on the project

Parameters:

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

Returns:

Type Description
dict[Literal['dataset', 'project'], Union[Dataset, Project]]

Dictionary with the dataset and project objects (keys: dataset, project).

Raises:

Type Description
HTTPException(status_code=422_UNPROCESSABLE_ENTITY)

If the dataset is not associated to the project

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
async def _get_dataset_check_owner(
    *,
    project_id: int,
    dataset_id: int,
    user_id: int,
    db: AsyncSession,
) -> dict[Literal["dataset", "project"], Union[Dataset, Project]]:
    """
    Get a dataset and a project, after access control on the project

    Args:
        project_id:
        dataset_id:
        user_id:
        db:

    Returns:
        Dictionary with the dataset and project objects (keys: `dataset`,
            `project`).

    Raises:
        HTTPException(status_code=422_UNPROCESSABLE_ENTITY):
            If the dataset is not associated to the project
    """

    # Access control for project
    project = await _get_project_check_owner(
        project_id=project_id, user_id=user_id, db=db
    )
    # Get dataset
    dataset = await db.get(Dataset, dataset_id)
    if not dataset:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Dataset not found"
        )
    if dataset.project_id != project_id:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Invalid {project_id=} for {dataset_id=}",
        )

    # Refresh so that dataset.project relationship is loaded (see discussion
    # in issue #1063)
    await db.refresh(dataset)

    return dict(dataset=dataset, project=project)

_get_job_check_owner(*, project_id, job_id, user_id, db) async

Get a job and a project, after access control on the project

Parameters:

Name Type Description Default
project_id int
required
job_id int
required
user_id int
required
db AsyncSession
required

Returns:

Type Description
dict[Literal['job', 'project'], Union[ApplyWorkflow, Project]]

Dictionary with the job and project objects (keys: job, project).

Raises:

Type Description
HTTPException(status_code=422_UNPROCESSABLE_ENTITY)

If the job is not associated to the project

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
async def _get_job_check_owner(
    *,
    project_id: int,
    job_id: int,
    user_id: int,
    db: AsyncSession,
) -> dict[Literal["job", "project"], Union[ApplyWorkflow, Project]]:
    """
    Get a job and a project, after access control on the project

    Args:
        project_id:
        job_id:
        user_id:
        db:

    Returns:
        Dictionary with the job and project objects (keys: `job`,
            `project`).

    Raises:
        HTTPException(status_code=422_UNPROCESSABLE_ENTITY):
            If the job is not associated to the project
    """
    # Access control for project
    project = await _get_project_check_owner(
        project_id=project_id, user_id=user_id, db=db
    )
    # Get dataset
    job = await db.get(ApplyWorkflow, job_id)
    if not job:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Job not found"
        )
    if job.project_id != project_id:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Invalid {project_id=} for {job_id=}",
        )
    return dict(job=job, project=project)

_get_project_check_owner(*, project_id, user_id, db) async

Check that user is a member of project and return the project.

Parameters:

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

Returns:

Type Description
Project

The project object

Raises:

Type Description
HTTPException(status_code=403_FORBIDDEN)

If the user is not a member of the project

HTTPException(status_code=404_NOT_FOUND)

If the project does not exist

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
async def _get_project_check_owner(
    *,
    project_id: int,
    user_id: int,
    db: AsyncSession,
) -> Project:
    """
    Check that user is a member of project and return the project.

    Args:
        project_id:
        user_id:
        db:

    Returns:
        The project object

    Raises:
        HTTPException(status_code=403_FORBIDDEN):
            If the user is not a member of the project
        HTTPException(status_code=404_NOT_FOUND):
            If the project does not exist
    """
    project = await db.get(Project, project_id)
    link_user_project = await db.get(LinkUserProject, (project_id, user_id))
    if not project:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Project not found"
        )
    if not link_user_project:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=f"Not allowed on project {project_id}",
        )
    return project

_get_submitted_jobs_statement()

Returns:

Type Description
SelectOfScalar

A sqlmodel statement that selects all ApplyWorkflows with

SelectOfScalar

ApplyWorkflow.status equal to submitted.

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
387
388
389
390
391
392
393
394
395
396
def _get_submitted_jobs_statement() -> SelectOfScalar:
    """
    Returns:
        A sqlmodel statement that selects all `ApplyWorkflow`s with
        `ApplyWorkflow.status` equal to `submitted`.
    """
    stm = select(ApplyWorkflow).where(
        ApplyWorkflow.status == JobStatusTypeV1.SUBMITTED
    )
    return stm

_get_task_check_owner(*, task_id, user, db) async

Get a task, after access control.

This check constitutes a preliminary version of access control: if the current user is not a superuser and differs from the task owner (including when owner is None), we raise an 403 HTTP Exception.

Parameters:

Name Type Description Default
task_id int
required
user UserOAuth
required
db AsyncSession
required

Returns:

Type Description
Task

The task object.

Raises:

Type Description
HTTPException(status_code=404_NOT_FOUND)

If the task does not exist

HTTPException(status_code=403_FORBIDDEN)

If the user does not have rights to edit this task.

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
async def _get_task_check_owner(
    *,
    task_id: int,
    user: UserOAuth,
    db: AsyncSession,
) -> Task:
    """
    Get a task, after access control.

    This check constitutes a preliminary version of access control:
    if the current user is not a superuser and differs from the task owner
    (including when `owner is None`), we raise an 403 HTTP Exception.

    Args:
        task_id:
        user:
        db:

    Returns:
        The task object.

    Raises:
        HTTPException(status_code=404_NOT_FOUND):
            If the task does not exist
        HTTPException(status_code=403_FORBIDDEN):
            If the user does not have rights to edit this task.
    """
    task = await db.get(Task, task_id)
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task {task_id} not found.",
        )

    if not user.is_superuser:
        if task.owner is None:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=(
                    "Only a superuser can modify a Task with `owner=None`."
                ),
            )
        else:
            if user.username:
                owner = user.username
            else:
                verify_user_has_settings(user)
                owner = user.settings.slurm_user
            if owner != task.owner:
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail=(
                        f"Current user ({owner}) cannot modify Task {task.id} "
                        f"with different owner ({task.owner})."
                    ),
                )
    return task

_get_workflow_check_owner(*, workflow_id, project_id, user_id, db) async

Get a workflow and a project, after access control on the project.

Parameters:

Name Type Description Default
workflow_id int
required
project_id int
required
user_id int
required
db AsyncSession
required

Returns:

Type Description
Workflow

The workflow object.

Raises:

Type Description
HTTPException(status_code=404_NOT_FOUND)

If the workflow does not exist

HTTPException(status_code=422_UNPROCESSABLE_ENTITY)

If the workflow is not associated to the project

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
async def _get_workflow_check_owner(
    *,
    workflow_id: int,
    project_id: int,
    user_id: int,
    db: AsyncSession,
) -> Workflow:
    """
    Get a workflow and a project, after access control on the project.

    Args:
        workflow_id:
        project_id:
        user_id:
        db:

    Returns:
        The workflow object.

    Raises:
        HTTPException(status_code=404_NOT_FOUND):
            If the workflow does not exist
        HTTPException(status_code=422_UNPROCESSABLE_ENTITY):
            If the workflow is not associated to the project
    """

    # Access control for project
    project = await _get_project_check_owner(
        project_id=project_id, user_id=user_id, db=db
    )
    # Get workflow
    workflow = await db.get(Workflow, workflow_id)
    if not workflow:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Workflow not found"
        )
    if workflow.project_id != project.id:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=(f"Invalid {project_id=} for {workflow_id=}."),
        )

    # Refresh so that workflow.project relationship is loaded (see discussion
    # in issue #1063)
    await db.refresh(workflow)

    return workflow

_get_workflow_task_check_owner(*, project_id, workflow_id, workflow_task_id, user_id, db) async

Check that user has access to Workflow and WorkflowTask.

Parameters:

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

Returns:

Type Description
tuple[WorkflowTask, Workflow]

Tuple of WorkflowTask and Workflow objects.

Raises:

Type Description
HTTPException(status_code=404_NOT_FOUND)

If the WorkflowTask does not exist

HTTPException(status_code=422_UNPROCESSABLE_ENTITY)

If the WorkflowTask is not associated to the Workflow

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
async def _get_workflow_task_check_owner(
    *,
    project_id: int,
    workflow_id: int,
    workflow_task_id: int,
    user_id: int,
    db: AsyncSession,
) -> tuple[WorkflowTask, Workflow]:
    """
    Check that user has access to Workflow and WorkflowTask.

    Args:
        project_id:
        workflow_id:
        workflow_task_id:
        user_id:
        db:

    Returns:
        Tuple of WorkflowTask and Workflow objects.

    Raises:
        HTTPException(status_code=404_NOT_FOUND):
            If the WorkflowTask does not exist
        HTTPException(status_code=422_UNPROCESSABLE_ENTITY):
            If the WorkflowTask is not associated to the Workflow
    """

    # Access control for workflow
    workflow = await _get_workflow_check_owner(
        workflow_id=workflow_id, project_id=project_id, user_id=user_id, db=db
    )

    # If WorkflowTask is not in the db, exit
    workflow_task = await db.get(WorkflowTask, workflow_task_id)
    if not workflow_task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="WorkflowTask not found",
        )

    # If WorkflowTask is not part of the expected Workflow, exit
    if workflow_id != workflow_task.workflow_id:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"Invalid {workflow_id=} for {workflow_task_id=}",
        )

    return workflow_task, workflow

_workflow_insert_task(*, workflow_id, task_id, args=None, meta=None, order=None, db) async

Insert a new WorkflowTask into Workflow.task_list

Parameters:

Name Type Description Default
task_id int

TBD

required
args Optional[dict[str, Any]]

TBD

None
meta Optional[dict[str, Any]]

TBD

None
order Optional[int]

TBD

None
db AsyncSession

TBD

required
Source code in fractal_server/app/routes/api/v1/_aux_functions.py
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
async def _workflow_insert_task(
    *,
    workflow_id: int,
    task_id: int,
    args: Optional[dict[str, Any]] = None,
    meta: Optional[dict[str, Any]] = None,
    order: Optional[int] = None,
    db: AsyncSession,
) -> WorkflowTask:
    """
    Insert a new WorkflowTask into Workflow.task_list

    Args:
        task_id: TBD
        args: TBD
        meta: TBD
        order: TBD
        db: TBD
    """
    db_workflow = await db.get(Workflow, workflow_id)
    if db_workflow is None:
        raise ValueError(f"Workflow {workflow_id} does not exist")

    if order is None:
        order = len(db_workflow.task_list)

    # Get task from db, and extract default arguments via a Task property
    # method
    db_task = await db.get(Task, task_id)
    if db_task is None:
        raise ValueError(f"Task {task_id} does not exist")

    default_args = db_task.default_args_from_args_schema
    # Override default_args with args
    actual_args = default_args.copy()
    if args is not None:
        for k, v in args.items():
            actual_args[k] = v
    if not actual_args:
        actual_args = None

    # Combine meta (higher priority) and db_task.meta (lower priority)
    wt_meta = (db_task.meta or {}).copy()
    wt_meta.update(meta or {})
    if not wt_meta:
        wt_meta = None

    # Create DB entry
    wf_task = WorkflowTask(task_id=task_id, args=actual_args, meta=wt_meta)
    db.add(wf_task)
    db_workflow.task_list.insert(order, wf_task)
    db_workflow.task_list.reorder()  # type: ignore
    await db.commit()
    await db.refresh(wf_task)

    return wf_task

clean_app_job_list_v1(db, jobs_list) async

Remove from a job list all jobs with status different from submitted.

Parameters:

Name Type Description Default
db AsyncSession

Async database session

required
jobs_list list[int]

List of job IDs currently associated to the app.

required
Return

List of IDs for submitted jobs.

Source code in fractal_server/app/routes/api/v1/_aux_functions.py
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
async def clean_app_job_list_v1(
    db: AsyncSession, jobs_list: list[int]
) -> list[int]:
    """
    Remove from a job list all jobs with status different from submitted.

    Args:
        db: Async database session
        jobs_list: List of job IDs currently associated to the app.

    Return:
        List of IDs for submitted jobs.
    """
    stmt = select(ApplyWorkflow).where(ApplyWorkflow.id.in_(jobs_list))
    result = await db.execute(stmt)
    db_jobs_list = result.scalars().all()
    submitted_job_ids = [
        job.id
        for job in db_jobs_list
        if job.status == JobStatusTypeV1.SUBMITTED
    ]
    return submitted_job_ids