Skip to content

_aux_sync_core_tasks

Attributes

Classes

_CoreInfoSet

Bases: RootModel

Set of (pkg_name, version, task_name) tuples.

Source code in fractal_server/cli/_aux_sync_core_tasks.py 
20
21
22
23
24
25
class _CoreInfoSet(RootModel):
    """
    Set of `(pkg_name, version, task_name)` tuples.
    """

    root: _SetThreeStringsTuple

Functions:

_count_core_tasks(db_sync)

Count core tasks.

Source code in fractal_server/cli/_aux_sync_core_tasks.py 
49
50
51
52
53
54
55
def _count_core_tasks(db_sync: Session) -> int:
    """
    Count core tasks.
    """
    count_stm = select(func.count(TaskV2.id)).where(TaskV2.is_core.is_(True))
    res = db_sync.execute(count_stm)
    return res.scalar_one()

_read_set_from_file(path)

Read a file (if any) and parse into a set of core-task info items.

Source code in fractal_server/cli/_aux_sync_core_tasks.py 
28
29
30
31
32
33
def _read_set_from_file(path: Path | None) -> _SetThreeStringsTuple:
    """
    Read a file (if any) and parse into a set of core-task info items.
    """
    json_data = path.read_text() if path else "[]"
    return _CoreInfoSet.model_validate_json(json_data).root