Skip to content

task_group

CLASS DESCRIPTION
TaskGroupV2

Classes

TaskGroupV2

Bases: Base

Returned by: Used by:
ATTRIBUTE DESCRIPTION
pinned_package_versions_post_string

Prepare string to be used in python -m pip install.

TYPE: str

pinned_package_versions_pre_string

Prepare string to be used in python -m pip install.

TYPE: str

pip_install_string

Prepare string to be used in python -m pip install.

TYPE: str

Source code in fractal_server/app/models/v2/task_group.py 
 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
 73
 74
 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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class TaskGroupV2(Base):
    __tablename__ = "taskgroupv2"

    id: Mapped[int] = mapped_column(primary_key=True)
    task_list: Mapped[list["TaskV2"]] = relationship(
        lazy="selectin", cascade="all, delete-orphan"
    )

    user_id: Mapped[int] = mapped_column(ForeignKey("user_oauth.id"))
    user_group_id: Mapped[int | None] = mapped_column(
        ForeignKey("usergroup.id", ondelete="SET NULL"), default=lambda: None
    )
    resource_id: Mapped[int] = mapped_column(
        ForeignKey("resource.id", ondelete="RESTRICT")
    )

    origin: Mapped[str]
    pkg_name: Mapped[str]
    version: Mapped[str]
    python_version: Mapped[str | None] = mapped_column(default=lambda: None)
    pixi_version: Mapped[str | None] = mapped_column(default=lambda: None)
    path: Mapped[str | None] = mapped_column(default=lambda: None)
    archive_path: Mapped[str | None] = mapped_column(default=lambda: None)
    pip_extras: Mapped[str | None] = mapped_column(default=lambda: None)
    pinned_package_versions_pre: Mapped[dict[str, str]] = mapped_column(
        JSONB,
        server_default="{}",
        default={},
        nullable=True,
    )
    pinned_package_versions_post: Mapped[dict[str, str]] = mapped_column(
        JSONB,
        server_default="{}",
        default={},
        nullable=True,
    )
    env_info: Mapped[str | None] = mapped_column(default=lambda: None)
    venv_path: Mapped[str | None] = mapped_column(default=lambda: None)

    active: Mapped[bool] = mapped_column(default=True)
    timestamp_created: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=get_timestamp
    )
    timestamp_last_used: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=(
            datetime(2024, 11, 20, tzinfo=timezone.utc).isoformat()
        ),
        default=get_timestamp,
    )

    __table_args__ = (
        Index(
            "ix_taskgroupv2_path_unique_constraint",
            "path",
            "resource_id",
            unique=True,
        ),
    )

    @property
    def pip_install_string(self) -> str:
        """
        Prepare string to be used in `python -m pip install`.
        """
        _check_origin_not_pixi(self.origin)

        extras = (
            f"[{self.pip_extras}]"
            if (self.pip_extras is not None and self.pip_extras != "")
            else ""
        )

        if self.archive_path is not None:
            return f"{self.archive_path}{extras}"
        else:
            return f"{self.pkg_name}{extras}=={self.version}"

    @property
    def pinned_package_versions_pre_string(self) -> str:
        """
        Prepare string to be used in `python -m pip install`.
        """
        _check_origin_not_pixi(self.origin)

        if self.pinned_package_versions_pre is None:
            return ""
        output = _create_dependency_string(self.pinned_package_versions_pre)
        return output

    @property
    def pinned_package_versions_post_string(self) -> str:
        """
        Prepare string to be used in `python -m pip install`.
        """
        _check_origin_not_pixi(self.origin)

        if self.pinned_package_versions_post is None:
            return ""
        output = _create_dependency_string(self.pinned_package_versions_post)
        return output

Attributes

pinned_package_versions_post_string property

Prepare string to be used in python -m pip install.

pinned_package_versions_pre_string property

Prepare string to be used in python -m pip install.

pip_install_string property

Prepare string to be used in python -m pip install.

Functions:

_check_origin_not_pixi(origin)

Raise ValueError if origin=="pixi"

Source code in fractal_server/app/models/v2/task_group.py 
21
22
23
24
25
26
def _check_origin_not_pixi(origin: str) -> None:
    """
    Raise `ValueError` if `origin=="pixi"`
    """
    if origin == "pixi":
        raise ValueError(f"Cannot call 'pip_install_string' if {origin=}.")

_create_dependency_string(pinned_versions)

Expand e.g. {"a": "1.2", "b": "3"} into '"a==1.2" "b==3"'.

Source code in fractal_server/app/models/v2/task_group.py 
29
30
31
32
33
34
35
36
def _create_dependency_string(pinned_versions: dict[str, str]) -> str:
    """
    Expand e.g. `{"a": "1.2", "b": "3"}` into `'"a==1.2" "b==3"'`.
    """
    output = " ".join(
        [f'"{key}=={value}"' for key, value in pinned_versions.items()]
    )
    return output