Skip to content

Utils database

FUNCTION DESCRIPTION
create_db_tasks_and_update_task_group_async

Create a TaskGroupV2 with N TaskV2s, and insert them into the database.
create_db_tasks_and_update_task_group_sync

Create a TaskGroupV2 with N TaskV2s, and insert them into the database.

Classes

Functions:

create_db_tasks_and_update_task_group_async(*, task_group_id, task_list, db) async

Create a TaskGroupV2 with N TaskV2s, and insert them into the database.

PARAMETER DESCRIPTION

task_group_id

ID of an existing TaskGroupV2 object.

TYPE: int

task_list

List of TaskCreate objects to be inserted into the db.

TYPE: list[TaskCreate]

db

Synchronous database session

TYPE: AsyncSession

RETURNS DESCRIPTION
TaskGroupV2

Updated TaskGroupV2 object.
Source code in fractal_server/tasks/v2/utils_database.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def create_db_tasks_and_update_task_group_async(
    *,
    task_group_id: int,
    task_list: list[TaskCreate],
    db: AsyncSession,
) -> TaskGroupV2:
    """
    Create a `TaskGroupV2` with N `TaskV2`s, and insert them into the database.

    Args:
        task_group_id: ID of an existing `TaskGroupV2` object.
        task_list: List of `TaskCreate` objects to be inserted into the db.
        db: Synchronous database session

    Returns:
        Updated `TaskGroupV2` object.
    """
    actual_task_list = [TaskV2(**task.model_dump()) for task in task_list]
    task_group = await db.get_one(TaskGroupV2, task_group_id)
    task_group.task_list = actual_task_list
    db.add(task_group)
    await db.commit()
    await db.refresh(task_group)

    return task_group

create_db_tasks_and_update_task_group_sync(*, task_group_id, task_list, db)

Create a TaskGroupV2 with N TaskV2s, and insert them into the database.

PARAMETER DESCRIPTION

task_group_id

ID of an existing TaskGroupV2 object.

TYPE: int

task_list

List of TaskCreate objects to be inserted into the db.

TYPE: list[TaskCreate]

db

Synchronous database session

TYPE: Session

RETURNS DESCRIPTION
TaskGroupV2

Updated TaskGroupV2 object.
Source code in fractal_server/tasks/v2/utils_database.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def create_db_tasks_and_update_task_group_sync(
    *,
    task_group_id: int,
    task_list: list[TaskCreate],
    db: DBSyncSession,
) -> TaskGroupV2:
    """
    Create a `TaskGroupV2` with N `TaskV2`s, and insert them into the database.

    Args:
        task_group_id: ID of an existing `TaskGroupV2` object.
        task_list: List of `TaskCreate` objects to be inserted into the db.
        db: Synchronous database session

    Returns:
        Updated `TaskGroupV2` object.
    """
    actual_task_list = [TaskV2(**task.model_dump()) for task in task_list]
    task_group = db.get_one(TaskGroupV2, task_group_id)
    task_group.task_list = actual_task_list
    db.add(task_group)
    db.commit()
    db.refresh(task_group)

    return task_group