26644807005

Committed 29 May 2026 03:01PM UTC coverage: 86.375% (+0.07%) from 86.308%

Build # 26644807005

Build Type

Pull #1315

github

Committed by

web-flow

Commit Message

Merge 2ac8da5e6 into 1c50f11c2

Pull Request Pull Request #1315: feat: resource pools access through integration

Coverage Stats

26829 of 31061 relevant lines covered (86.38%)

3.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.61

/components/renku_data_services/session/models.py

"""Models for sessions."""

from __future__ import annotations

import typing
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import StrEnum
from pathlib import PurePosixPath
from typing import TYPE_CHECKING, Final

from ulid import ULID

from renku_data_services import errors
from renku_data_services.base_models.core import ResetType
from renku_data_services.session import crs

if TYPE_CHECKING:
    from renku_data_services.session import apispec, config

from .constants import (
    BUILD_GID,
    BUILD_MOUNT_DIRECTORY,
    BUILD_PORT,
    BUILD_UID,
    BUILD_WORKING_DIRECTORY,
    ENV_VARIABLE_NAME_MATCHER,
    ENV_VARIABLE_REGEX,
)


@dataclass(frozen=True, eq=True, kw_only=True)
class Member:
    """Member model."""

    id: str


class EnvironmentKind(StrEnum):
    """The type of environment."""

    GLOBAL = "GLOBAL"
    CUSTOM = "CUSTOM"


class EnvironmentImageSource(StrEnum):
    """The source of the environment image."""

    image = "image"
    build = "build"


class Platform(StrEnum):
    """The runtime platform."""

    linux_amd64 = "linux/amd64"
    linux_arm64 = "linux/arm64"


class BuilderVariant(StrEnum):
    """The type of environment builder."""

    python = "python"
    r = "r"


class FrontendVariant(StrEnum):
    """The environment frontend choice."""

    vscodium = "vscodium"
    jupyterlab = "jupyterlab"
    ttyd = "ttyd"
    rstudio = "rstudio"


VALID_BUILDER_FRONTEND_COMBINATIONS: typing.Final[set[tuple[BuilderVariant, FrontendVariant]]] = {
    (BuilderVariant.r, FrontendVariant.rstudio),
    (BuilderVariant.python, FrontendVariant.vscodium),
    (BuilderVariant.python, FrontendVariant.jupyterlab),
    (BuilderVariant.python, FrontendVariant.ttyd),
}


@dataclass(kw_only=True, frozen=True, eq=True)
class UnsavedBuildParameters:
    """The parameters of a build."""

    repository: str
    platforms: list[Platform]
    builder_variant: str
    frontend_variant: str
    repository_revision: str | None = None
    context_dir: str | None = None


@dataclass(kw_only=True, frozen=True, eq=True)
class BuildParameters(UnsavedBuildParameters):
    """BuildParameters saved in the database."""

    id: ULID


@dataclass(kw_only=True, frozen=True, eq=True)
class UnsavedEnvironment:
    """Session environment model that has not been saved."""

    name: str
    description: str | None = None
    container_image: str
    default_url: str
    port: int = 8888
    working_directory: PurePosixPath | None = None
    mount_directory: PurePosixPath | None = None
    uid: int = 1000
    gid: int = 1000
    environment_kind: EnvironmentKind
    environment_image_source: EnvironmentImageSource
    args: list[str] | None = None
    command: list[str] | None = None
    is_archived: bool = False
    strip_path_prefix: bool = False

    def __post_init__(self) -> None:
        if self.working_directory and not self.working_directory.is_absolute():
            raise errors.ValidationError(message="The working directory for a session is supposed to be absolute")
        if self.mount_directory and not self.mount_directory.is_absolute():
            raise errors.ValidationError(message="The mount directory for a session is supposed to be absolute")
        if self.working_directory and self.working_directory.is_reserved():
            raise errors.ValidationError(
                message="The requested value for the working directory is reserved by the OS and cannot be used."
            )
        if self.mount_directory and self.mount_directory.is_reserved():
            raise errors.ValidationError(
                message="The requested value for the mount directory is reserved by the OS and cannot be used."
            )


@dataclass(frozen=True, eq=True, kw_only=True)
class Environment(UnsavedEnvironment):
    """Session environment model."""

    id: ULID
    creation_date: datetime
    created_by: Member
    container_image: str
    default_url: str
    port: int
    working_directory: PurePosixPath | None
    mount_directory: PurePosixPath | None
    uid: int
    gid: int
    build_parameters: BuildParameters | None
    build_parameters_id: ULID | None


@dataclass(kw_only=True, frozen=True, eq=True)
class BuildParametersPatch:
    """Patch for parameters of a build."""

    repository: str | None = None
    platforms: list[Platform] | None = None
    builder_variant: str | None = None
    frontend_variant: str | None = None
    repository_revision: str | None = None
    context_dir: str | None = None


@dataclass(eq=True, kw_only=True)
class EnvironmentPatch:
    """Model for changes requested on a session environment."""

    name: str | None = None
    description: str | None = None
    container_image: str | None = None
    default_url: str | None = None
    port: int | None = None
    working_directory: PurePosixPath | ResetType | None = None
    mount_directory: PurePosixPath | ResetType | None = None
    uid: int | None = None
    gid: int | None = None
    args: list[str] | ResetType | None = None
    command: list[str] | ResetType | None = None
    is_archived: bool | None = None
    build_parameters: BuildParametersPatch | None = None
    environment_image_source: EnvironmentImageSource | None = None
    strip_path_prefix: bool | None = None


# TODO: Verify that these limits are compatible with k8s
MAX_NUMBER_ENV_VARIABLES: typing.Final[int] = 32
MAX_LENGTH_ENV_VARIABLES_NAME: typing.Final[int] = 256
MAX_LENGTH_ENV_VARIABLES_VALUE: typing.Final[int] = 1000


@dataclass(frozen=True, eq=True, kw_only=True)
class EnvVar:
    """Model for an environment variable."""

    name: str
    value: str | None = None

    @classmethod
    def from_dict(cls, env_dict: dict[str, str | None]) -> list[EnvVar]:
        """Create a list of EnvVar instances from a dictionary."""
        return [cls(name=name, value=value) for name, value in env_dict.items()]

    @classmethod
    def from_apispec(cls, env_variables: list[apispec.EnvVar]) -> list[EnvVar]:
        """Create a list of EnvVar instances from apispec objects."""
        return [cls(name=env_var.name, value=env_var.value) for env_var in env_variables]

    @classmethod
    def to_dict(cls, env_variables: list[EnvVar]) -> dict[str, str | None]:
        """Convert to dict."""
        return {var.name: var.value for var in env_variables}

    def __post_init__(self) -> None:
        error_msgs: list[str] = []
        if len(self.name) > MAX_LENGTH_ENV_VARIABLES_NAME:
            error_msgs.append(
                f"Env variable name '{self.name}' is longer than {MAX_LENGTH_ENV_VARIABLES_NAME} characters."
            )
        if self.name.upper().startswith("RENKU"):
            error_msgs.append(f"Env variable name '{self.name}' should not start with 'RENKU'.")
        if ENV_VARIABLE_NAME_MATCHER.match(self.name) is None:
            error_msgs.append(f"Env variable name '{self.name}' must match the regex '{ENV_VARIABLE_REGEX}'.")
        if self.value and len(self.value) > MAX_LENGTH_ENV_VARIABLES_VALUE:
            error_msgs.append(
                f"Env variable value for '{self.name}' is longer than {MAX_LENGTH_ENV_VARIABLES_VALUE} characters."
            )

        if error_msgs:
            if len(error_msgs) == 1:
                raise errors.ValidationError(message=error_msgs[0])
            raise errors.ValidationError(message="\n".join(error_msgs))


class LauncherType(StrEnum):
    """The launcher type."""

    interactive = "interactive"
    non_interactive = "non_interactive"
    app = "app"


@dataclass(frozen=True, eq=True, kw_only=True)
class UnsavedSessionLauncher:
    """Session launcher model that has not been persisted in the DB."""

    project_id: ULID
    name: str
    description: str | None
    resource_class_id: int | None
    disk_storage: int | None
    env_variables: list[EnvVar] | None
    environment: str | UnsavedEnvironment | UnsavedBuildParameters
    """When a string is passed for the environment it should be the ID of an existing environment."""
    launcher_type: LauncherType


@dataclass(frozen=True, eq=True, kw_only=True)
class SessionLauncher(UnsavedSessionLauncher):
    """Session launcher model."""

    id: ULID
    creation_date: datetime
    created_by: Member
    environment: Environment


@dataclass(frozen=True, eq=True, kw_only=True)
class SessionLauncherPatch:
    """Model for changes requested on a session launcher."""

    name: str | None = None
    description: str | None = None
    # NOTE: When unsaved environment is used it means a brand-new environment should be created for the
    # launcher with the update of the launcher.
    environment: str | EnvironmentPatch | UnsavedEnvironment | UnsavedBuildParameters | None = None
    resource_class_id: int | None | ResetType = None
    disk_storage: int | None | ResetType = None
    env_variables: list[EnvVar] | None | ResetType = None


@dataclass(frozen=True, eq=True, kw_only=True)
class BuildResult:
    """Model to represent the result of a build of a container image."""

    image: str
    completed_at: datetime
    repository_url: str
    repository_git_commit_sha: str


class BuildStatus(StrEnum):
    """The status of a build."""

    in_progress = "in_progress"
    failed = "failed"
    cancelled = "cancelled"
    succeeded = "succeeded"


@dataclass(frozen=True, eq=True, kw_only=True)
class Build:
    """Model to represent the build of a container image."""

    id: ULID
    environment_id: ULID
    created_at: datetime
    status: BuildStatus
    result: BuildResult | None = None
    error_reason: str | None = None

    @property
    def k8s_name(self) -> str:
        """Returns the name of the corresponding Shipwright BuildRun."""
        name = f"renku-{self.id}"
        return name.lower()


@dataclass(frozen=True, eq=True, kw_only=True)
class UnsavedBuild:
    """Model to represent a requested container image build."""

    environment_id: ULID


@dataclass(frozen=True, eq=True, kw_only=True)
class BuildPatch:
    """Model to represent the requested update to a container image build."""

    status: BuildStatus | None = None


@dataclass(frozen=True, eq=True, kw_only=True)
class AuthenticationSecret:
    """Model to represent the secret to be created/used for private repositories."""

    username: str
    password: str

    def to_string_data(self) -> dict:
        """Returns the raw data as is to be used as Secret stringData."""

        return {
            "username": self.username,
            "password": self.password,
        }


@dataclass(frozen=True, eq=True, kw_only=True)
class ShipwrightBuildRunParams:
    """Model to represent the parameters used to create a new Shipwright BuildRun."""

    name: str
    git_repository: str
    run_image: str
    output_image: str
    build_strategy_name: str
    push_secret_name: str
    authentication_secret: AuthenticationSecret | None = None
    retention_after_failed: timedelta | None = None
    retention_after_succeeded: timedelta | None = None
    build_timeout: timedelta | None = None
    node_selector: dict[str, str] | None = None
    tolerations: list[crs.Toleration] | None = None
    labels: dict[str, str] | None = None
    annotations: dict[str, str] | None = None
    frontend: str = FrontendVariant.vscodium.value
    builder_image: str | None = None
    git_repository_revision: str | None = None
    context_dir: str | None = None
    insecure_registries: str = ""

    def with_overrides(self, overrides: config.BuildPlatformOverrides | None) -> ShipwrightBuildRunParams:
        """Returns a copy of the BuildRun parameters with overrides applied."""
        if overrides is None:
            return self
        return ShipwrightBuildRunParams(
            name=self.name,
            git_repository=self.git_repository,
            run_image=overrides.run_image or self.run_image,
            output_image=self.output_image,
            build_strategy_name=overrides.strategy_name or self.build_strategy_name,
            authentication_secret=self.authentication_secret,
            push_secret_name=self.push_secret_name,
            retention_after_failed=self.retention_after_failed,
            retention_after_succeeded=self.retention_after_succeeded,
            build_timeout=self.build_timeout,
            node_selector=overrides.node_selector or self.node_selector,
            tolerations=overrides.tolerations or self.tolerations,
            labels=self.labels,
            annotations=self.annotations,
            frontend=self.frontend,
            builder_image=overrides.builder_image or self.builder_image,
            git_repository_revision=self.git_repository_revision,
            context_dir=self.context_dir,
            insecure_registries=self.insecure_registries,
        )


@dataclass(frozen=True, eq=True, kw_only=True)
class ShipwrightBuildStatusUpdateContent:
    """Model to represent an update about a build from Shipwright."""

    status: BuildStatus
    result: BuildResult | None = None
    completed_at: datetime | None = None
    error_reason: str | None = None


@dataclass(frozen=True, eq=True, kw_only=True)
class ShipwrightBuildStatusUpdate:
    """Model to represent an update about a build from Shipwright."""

    update: ShipwrightBuildStatusUpdateContent | None
    """The update about a build.

    None represents "no update"."""


BUILD_ENVIRONMENT_CONFIGS: Final[dict[str, UnsavedEnvironment]] = {
    FrontendVariant.rstudio.value: UnsavedEnvironment(
        name="rstudio",
        default_url="/",
        port=BUILD_PORT,
        container_image="image:unknown-at-the-moment",
        working_directory=BUILD_WORKING_DIRECTORY,
        mount_directory=BUILD_MOUNT_DIRECTORY,
        uid=BUILD_UID,
        gid=BUILD_GID,
        environment_kind=EnvironmentKind.CUSTOM,
        environment_image_source=EnvironmentImageSource.build,
        strip_path_prefix=True,
    ),
    FrontendVariant.jupyterlab.value: UnsavedEnvironment(
        name="jupyterlab",
        default_url="/lab",
        port=BUILD_PORT,
        container_image="image:unknown-at-the-moment",
        working_directory=BUILD_WORKING_DIRECTORY,
        mount_directory=BUILD_MOUNT_DIRECTORY,
        uid=BUILD_UID,
        gid=BUILD_GID,
        environment_kind=EnvironmentKind.CUSTOM,
        environment_image_source=EnvironmentImageSource.build,
        strip_path_prefix=False,
    ),
    FrontendVariant.ttyd.value: UnsavedEnvironment(
        name="ttyd",
        default_url="/",
        port=BUILD_PORT,
        container_image="image:unknown-at-the-moment",
        working_directory=BUILD_WORKING_DIRECTORY,
        mount_directory=BUILD_MOUNT_DIRECTORY,
        uid=BUILD_UID,
        gid=BUILD_GID,
        environment_kind=EnvironmentKind.CUSTOM,
        environment_image_source=EnvironmentImageSource.build,
        strip_path_prefix=False,
    ),
    FrontendVariant.vscodium.value: UnsavedEnvironment(
        name="vscodium",
        default_url="/",
        port=BUILD_PORT,
        container_image="image:unknown-at-the-moment",
        working_directory=BUILD_WORKING_DIRECTORY,
        mount_directory=BUILD_MOUNT_DIRECTORY,
        uid=BUILD_UID,
        gid=BUILD_GID,
        environment_kind=EnvironmentKind.CUSTOM,
        environment_image_source=EnvironmentImageSource.build,
        strip_path_prefix=False,
    ),
}

1	"""Models for sessions."""
2
3	from __future__ import annotations	4✔
4
5	import typing	4✔
6	from dataclasses import dataclass	4✔
7	from datetime import datetime, timedelta	4✔
8	from enum import StrEnum	4✔
9	from pathlib import PurePosixPath	4✔
10	from typing import TYPE_CHECKING, Final	4✔
11
12	from ulid import ULID	4✔
13
14	from renku_data_services import errors	4✔
15	from renku_data_services.base_models.core import ResetType	4✔
16	from renku_data_services.session import crs	4✔
17
18	if TYPE_CHECKING:	4✔
19	from renku_data_services.session import apispec, config	×
20
21	from .constants import (	4✔
22	BUILD_GID,
23	BUILD_MOUNT_DIRECTORY,
24	BUILD_PORT,
25	BUILD_UID,
26	BUILD_WORKING_DIRECTORY,
27	ENV_VARIABLE_NAME_MATCHER,
28	ENV_VARIABLE_REGEX,
29	)
30
31
32	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
33	class Member:	4✔
34	"""Member model."""
35
36	id: str	4✔
37
38
39	class EnvironmentKind(StrEnum):	4✔
40	"""The type of environment."""
41
42	GLOBAL = "GLOBAL"	4✔
43	CUSTOM = "CUSTOM"	4✔
44
45
46	class EnvironmentImageSource(StrEnum):	4✔
47	"""The source of the environment image."""
48
49	image = "image"	4✔
50	build = "build"	4✔
51
52
53	class Platform(StrEnum):	4✔
54	"""The runtime platform."""
55
56	linux_amd64 = "linux/amd64"	4✔
57	linux_arm64 = "linux/arm64"	4✔
58
59
60	class BuilderVariant(StrEnum):	4✔
61	"""The type of environment builder."""
62
63	python = "python"	4✔
64	r = "r"	4✔
65
66
67	class FrontendVariant(StrEnum):	4✔
68	"""The environment frontend choice."""
69
70	vscodium = "vscodium"	4✔
71	jupyterlab = "jupyterlab"	4✔
72	ttyd = "ttyd"	4✔
73	rstudio = "rstudio"	4✔
74
75
76	VALID_BUILDER_FRONTEND_COMBINATIONS: typing.Final[set[tuple[BuilderVariant, FrontendVariant]]] = {	4✔
77	(BuilderVariant.r, FrontendVariant.rstudio),
78	(BuilderVariant.python, FrontendVariant.vscodium),
79	(BuilderVariant.python, FrontendVariant.jupyterlab),
80	(BuilderVariant.python, FrontendVariant.ttyd),
81	}
82
83
84	@dataclass(kw_only=True, frozen=True, eq=True)	4✔
85	class UnsavedBuildParameters:	4✔
86	"""The parameters of a build."""
87
88	repository: str	4✔
89	platforms: list[Platform]	4✔
90	builder_variant: str	4✔
91	frontend_variant: str	4✔
92	repository_revision: str \| None = None	4✔
93	context_dir: str \| None = None	4✔
94
95
96	@dataclass(kw_only=True, frozen=True, eq=True)	4✔
97	class BuildParameters(UnsavedBuildParameters):	4✔
98	"""BuildParameters saved in the database."""
99
100	id: ULID	4✔
101
102
103	@dataclass(kw_only=True, frozen=True, eq=True)	4✔
104	class UnsavedEnvironment:	4✔
105	"""Session environment model that has not been saved."""
106
107	name: str	4✔
108	description: str \| None = None	4✔
109	container_image: str	4✔
110	default_url: str	4✔
111	port: int = 8888	4✔
112	working_directory: PurePosixPath \| None = None	4✔
113	mount_directory: PurePosixPath \| None = None	4✔
114	uid: int = 1000	4✔
115	gid: int = 1000	4✔
116	environment_kind: EnvironmentKind	4✔
117	environment_image_source: EnvironmentImageSource	4✔
118	args: list[str] \| None = None	4✔
119	command: list[str] \| None = None	4✔
120	is_archived: bool = False	4✔
121	strip_path_prefix: bool = False	4✔
122
123	def __post_init__(self) -> None:	4✔
124	if self.working_directory and not self.working_directory.is_absolute():	4✔
125	raise errors.ValidationError(message="The working directory for a session is supposed to be absolute")	×
126	if self.mount_directory and not self.mount_directory.is_absolute():	4✔
127	raise errors.ValidationError(message="The mount directory for a session is supposed to be absolute")	1✔
128	if self.working_directory and self.working_directory.is_reserved():	4✔
129	raise errors.ValidationError(	×
130	message="The requested value for the working directory is reserved by the OS and cannot be used."
131	)
132	if self.mount_directory and self.mount_directory.is_reserved():	4✔
133	raise errors.ValidationError(	×
134	message="The requested value for the mount directory is reserved by the OS and cannot be used."
135	)
136
137
138	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
139	class Environment(UnsavedEnvironment):	4✔
140	"""Session environment model."""
141
142	id: ULID	4✔
143	creation_date: datetime	4✔
144	created_by: Member	4✔
145	container_image: str	4✔
146	default_url: str	4✔
147	port: int	4✔
148	working_directory: PurePosixPath \| None	4✔
149	mount_directory: PurePosixPath \| None	4✔
150	uid: int	4✔
151	gid: int	4✔
152	build_parameters: BuildParameters \| None	4✔
153	build_parameters_id: ULID \| None	4✔
154
155
156	@dataclass(kw_only=True, frozen=True, eq=True)	4✔
157	class BuildParametersPatch:	4✔
158	"""Patch for parameters of a build."""
159
160	repository: str \| None = None	4✔
161	platforms: list[Platform] \| None = None	4✔
162	builder_variant: str \| None = None	4✔
163	frontend_variant: str \| None = None	4✔
164	repository_revision: str \| None = None	4✔
165	context_dir: str \| None = None	4✔
166
167
168	@dataclass(eq=True, kw_only=True)	4✔
169	class EnvironmentPatch:	4✔
170	"""Model for changes requested on a session environment."""
171
172	name: str \| None = None	4✔
173	description: str \| None = None	4✔
174	container_image: str \| None = None	4✔
175	default_url: str \| None = None	4✔
176	port: int \| None = None	4✔
177	working_directory: PurePosixPath \| ResetType \| None = None	4✔
178	mount_directory: PurePosixPath \| ResetType \| None = None	4✔
179	uid: int \| None = None	4✔
180	gid: int \| None = None	4✔
181	args: list[str] \| ResetType \| None = None	4✔
182	command: list[str] \| ResetType \| None = None	4✔
183	is_archived: bool \| None = None	4✔
184	build_parameters: BuildParametersPatch \| None = None	4✔
185	environment_image_source: EnvironmentImageSource \| None = None	4✔
186	strip_path_prefix: bool \| None = None	4✔
187
188
189	# TODO: Verify that these limits are compatible with k8s
190	MAX_NUMBER_ENV_VARIABLES: typing.Final[int] = 32	4✔
191	MAX_LENGTH_ENV_VARIABLES_NAME: typing.Final[int] = 256	4✔
192	MAX_LENGTH_ENV_VARIABLES_VALUE: typing.Final[int] = 1000	4✔
193
194
195	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
196	class EnvVar:	4✔
197	"""Model for an environment variable."""
198
199	name: str	4✔
200	value: str \| None = None	4✔
201
202	@classmethod	4✔
203	def from_dict(cls, env_dict: dict[str, str \| None]) -> list[EnvVar]:	4✔
204	"""Create a list of EnvVar instances from a dictionary."""
205	return [cls(name=name, value=value) for name, value in env_dict.items()]	2✔
206
207	@classmethod	4✔
208	def from_apispec(cls, env_variables: list[apispec.EnvVar]) -> list[EnvVar]:	4✔
209	"""Create a list of EnvVar instances from apispec objects."""
210	return [cls(name=env_var.name, value=env_var.value) for env_var in env_variables]	4✔
211
212	@classmethod	4✔
213	def to_dict(cls, env_variables: list[EnvVar]) -> dict[str, str \| None]:	4✔
214	"""Convert to dict."""
215	return {var.name: var.value for var in env_variables}	2✔
216
217	def __post_init__(self) -> None:	4✔
218	error_msgs: list[str] = []	4✔
219	if len(self.name) > MAX_LENGTH_ENV_VARIABLES_NAME:	4✔
220	error_msgs.append(	×
221	f"Env variable name '{self.name}' is longer than {MAX_LENGTH_ENV_VARIABLES_NAME} characters."
222	)
223	if self.name.upper().startswith("RENKU"):	4✔
224	error_msgs.append(f"Env variable name '{self.name}' should not start with 'RENKU'.")	2✔
225	if ENV_VARIABLE_NAME_MATCHER.match(self.name) is None:	4✔
226	error_msgs.append(f"Env variable name '{self.name}' must match the regex '{ENV_VARIABLE_REGEX}'.")	2✔
227	if self.value and len(self.value) > MAX_LENGTH_ENV_VARIABLES_VALUE:	4✔
228	error_msgs.append(	×
229	f"Env variable value for '{self.name}' is longer than {MAX_LENGTH_ENV_VARIABLES_VALUE} characters."
230	)
231
232	if error_msgs:	4✔
233	if len(error_msgs) == 1:	2✔
234	raise errors.ValidationError(message=error_msgs[0])	2✔
235	raise errors.ValidationError(message="\n".join(error_msgs))	×
236
237
238	class LauncherType(StrEnum):	4✔
239	"""The launcher type."""
240
241	interactive = "interactive"	4✔
242	non_interactive = "non_interactive"	4✔
243	app = "app"	4✔
244
245
246	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
247	class UnsavedSessionLauncher:	4✔
248	"""Session launcher model that has not been persisted in the DB."""
249
250	project_id: ULID	4✔
251	name: str	4✔
252	description: str \| None	4✔
253	resource_class_id: int \| None	4✔
254	disk_storage: int \| None	4✔
255	env_variables: list[EnvVar] \| None	4✔
256	environment: str \| UnsavedEnvironment \| UnsavedBuildParameters	4✔
257	"""When a string is passed for the environment it should be the ID of an existing environment."""	4✔
258	launcher_type: LauncherType	4✔
259
260
261	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
262	class SessionLauncher(UnsavedSessionLauncher):	4✔
263	"""Session launcher model."""
264
265	id: ULID	4✔
266	creation_date: datetime	4✔
267	created_by: Member	4✔
268	environment: Environment	4✔
269
270
271	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
272	class SessionLauncherPatch:	4✔
273	"""Model for changes requested on a session launcher."""
274
275	name: str \| None = None	4✔
276	description: str \| None = None	4✔
277	# NOTE: When unsaved environment is used it means a brand-new environment should be created for the
278	# launcher with the update of the launcher.
279	environment: str \| EnvironmentPatch \| UnsavedEnvironment \| UnsavedBuildParameters \| None = None	4✔
280	resource_class_id: int \| None \| ResetType = None	4✔
281	disk_storage: int \| None \| ResetType = None	4✔
282	env_variables: list[EnvVar] \| None \| ResetType = None	4✔
283
284
285	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
286	class BuildResult:	4✔
287	"""Model to represent the result of a build of a container image."""
288
289	image: str	4✔
290	completed_at: datetime	4✔
291	repository_url: str	4✔
292	repository_git_commit_sha: str	4✔
293
294
295	class BuildStatus(StrEnum):	4✔
296	"""The status of a build."""
297
298	in_progress = "in_progress"	4✔
299	failed = "failed"	4✔
300	cancelled = "cancelled"	4✔
301	succeeded = "succeeded"	4✔
302
303
304	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
305	class Build:	4✔
306	"""Model to represent the build of a container image."""
307
308	id: ULID	4✔
309	environment_id: ULID	4✔
310	created_at: datetime	4✔
311	status: BuildStatus	4✔
312	result: BuildResult \| None = None	4✔
313	error_reason: str \| None = None	4✔
314
315	@property	4✔
316	def k8s_name(self) -> str:	4✔
317	"""Returns the name of the corresponding Shipwright BuildRun."""
318	name = f"renku-{self.id}"	×
319	return name.lower()	×
320
321
322	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
323	class UnsavedBuild:	4✔
324	"""Model to represent a requested container image build."""
325
326	environment_id: ULID	4✔
327
328
329	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
330	class BuildPatch:	4✔
331	"""Model to represent the requested update to a container image build."""
332
333	status: BuildStatus \| None = None	4✔
334
335
336	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
337	class AuthenticationSecret:	4✔
338	"""Model to represent the secret to be created/used for private repositories."""
339
340	username: str	4✔
341	password: str	4✔
342
343	def to_string_data(self) -> dict:	4✔
344	"""Returns the raw data as is to be used as Secret stringData."""
345
346	return {	×
347	"username": self.username,
348	"password": self.password,
349	}
350
351
352	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
353	class ShipwrightBuildRunParams:	4✔
354	"""Model to represent the parameters used to create a new Shipwright BuildRun."""
355
356	name: str	4✔
357	git_repository: str	4✔
358	run_image: str	4✔
359	output_image: str	4✔
360	build_strategy_name: str	4✔
361	push_secret_name: str	4✔
362	authentication_secret: AuthenticationSecret \| None = None	4✔
363	retention_after_failed: timedelta \| None = None	4✔
364	retention_after_succeeded: timedelta \| None = None	4✔
365	build_timeout: timedelta \| None = None	4✔
366	node_selector: dict[str, str] \| None = None	4✔
367	tolerations: list[crs.Toleration] \| None = None	4✔
368	labels: dict[str, str] \| None = None	4✔
369	annotations: dict[str, str] \| None = None	4✔
370	frontend: str = FrontendVariant.vscodium.value	4✔
371	builder_image: str \| None = None	4✔
372	git_repository_revision: str \| None = None	4✔
373	context_dir: str \| None = None	4✔
374	insecure_registries: str = ""	4✔
375
376	def with_overrides(self, overrides: config.BuildPlatformOverrides \| None) -> ShipwrightBuildRunParams:	4✔
377	"""Returns a copy of the BuildRun parameters with overrides applied."""
378	if overrides is None:	×
379	return self	×
380	return ShipwrightBuildRunParams(	×
381	name=self.name,
382	git_repository=self.git_repository,
383	run_image=overrides.run_image or self.run_image,
384	output_image=self.output_image,
385	build_strategy_name=overrides.strategy_name or self.build_strategy_name,
386	authentication_secret=self.authentication_secret,
387	push_secret_name=self.push_secret_name,
388	retention_after_failed=self.retention_after_failed,
389	retention_after_succeeded=self.retention_after_succeeded,
390	build_timeout=self.build_timeout,
391	node_selector=overrides.node_selector or self.node_selector,
392	tolerations=overrides.tolerations or self.tolerations,
393	labels=self.labels,
394	annotations=self.annotations,
395	frontend=self.frontend,
396	builder_image=overrides.builder_image or self.builder_image,
397	git_repository_revision=self.git_repository_revision,
398	context_dir=self.context_dir,
399	insecure_registries=self.insecure_registries,
400	)
401
402
403	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
404	class ShipwrightBuildStatusUpdateContent:	4✔
405	"""Model to represent an update about a build from Shipwright."""
406
407	status: BuildStatus	4✔
408	result: BuildResult \| None = None	4✔
409	completed_at: datetime \| None = None	4✔
410	error_reason: str \| None = None	4✔
411
412
413	@dataclass(frozen=True, eq=True, kw_only=True)	4✔
414	class ShipwrightBuildStatusUpdate:	4✔
415	"""Model to represent an update about a build from Shipwright."""
416
417	update: ShipwrightBuildStatusUpdateContent \| None	4✔
418	"""The update about a build.	4✔
419
420	None represents "no update"."""
421
422
423	BUILD_ENVIRONMENT_CONFIGS: Final[dict[str, UnsavedEnvironment]] = {	4✔
424	FrontendVariant.rstudio.value: UnsavedEnvironment(
425	name="rstudio",
426	default_url="/",
427	port=BUILD_PORT,
428	container_image="image:unknown-at-the-moment",
429	working_directory=BUILD_WORKING_DIRECTORY,
430	mount_directory=BUILD_MOUNT_DIRECTORY,
431	uid=BUILD_UID,
432	gid=BUILD_GID,
433	environment_kind=EnvironmentKind.CUSTOM,
434	environment_image_source=EnvironmentImageSource.build,
435	strip_path_prefix=True,
436	),
437	FrontendVariant.jupyterlab.value: UnsavedEnvironment(
438	name="jupyterlab",
439	default_url="/lab",
440	port=BUILD_PORT,
441	container_image="image:unknown-at-the-moment",
442	working_directory=BUILD_WORKING_DIRECTORY,
443	mount_directory=BUILD_MOUNT_DIRECTORY,
444	uid=BUILD_UID,
445	gid=BUILD_GID,
446	environment_kind=EnvironmentKind.CUSTOM,
447	environment_image_source=EnvironmentImageSource.build,
448	strip_path_prefix=False,
449	),
450	FrontendVariant.ttyd.value: UnsavedEnvironment(
451	name="ttyd",
452	default_url="/",
453	port=BUILD_PORT,
454	container_image="image:unknown-at-the-moment",
455	working_directory=BUILD_WORKING_DIRECTORY,
456	mount_directory=BUILD_MOUNT_DIRECTORY,
457	uid=BUILD_UID,
458	gid=BUILD_GID,
459	environment_kind=EnvironmentKind.CUSTOM,
460	environment_image_source=EnvironmentImageSource.build,
461	strip_path_prefix=False,
462	),
463	FrontendVariant.vscodium.value: UnsavedEnvironment(
464	name="vscodium",
465	default_url="/",
466	port=BUILD_PORT,
467	container_image="image:unknown-at-the-moment",
468	working_directory=BUILD_WORKING_DIRECTORY,
469	mount_directory=BUILD_MOUNT_DIRECTORY,
470	uid=BUILD_UID,
471	gid=BUILD_GID,
472	environment_kind=EnvironmentKind.CUSTOM,
473	environment_image_source=EnvironmentImageSource.build,
474	strip_path_prefix=False,
475	),
476	}

SwissDataScienceCenter / renku-data-services / 26644807005

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous