15844898279

Committed 24 Jun 2025 08:06AM UTC coverage: 86.98% (-0.02%) from 86.999%

Build # 15844898279

Build Type

Pull #903

github

Committed by

web-flow

Commit Message

Merge 62759d788 into 4f5366ae7

Pull Request Pull Request #903: feat: update to rclone v1.70.0+renku-1

Run Details

2 of 2 new or added lines in 2 files covered. (100.0%)

6 existing lines in 4 files now uncovered.

21859 of 25131 relevant lines covered (86.98%)

1.53 hits per line

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

81.22

/components/renku_data_services/crc/models.py

"""Domain models for the application."""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import asdict, dataclass, field
from enum import StrEnum
from typing import Any, Optional, Protocol
from uuid import uuid4

from ulid import ULID

from renku_data_services import errors
from renku_data_services.errors import ValidationError


class ResourcesProtocol(Protocol):
    """Used to represent resource values present in a resource class or quota."""

    @property
    def cpu(self) -> float:
        """Cpu in fractional cores."""
        ...

    @property
    def gpu(self) -> int:
        """Number of GPUs."""
        ...

    @property
    def memory(self) -> int:
        """Memory in gigabytes."""
        ...

    @property
    def max_storage(self) -> Optional[int]:
        """Maximum allowable storage in gigabytes."""
        ...


class ResourcesCompareMixin:
    """A mixin that adds comparison operator support on ResourceClasses and Quotas."""

    def __compare(
        self,
        other: ResourcesProtocol,
        compare_func: Callable[[int | float, int | float], bool],
    ) -> bool:
        results = [
            compare_func(self.cpu, other.cpu),  # type: ignore[attr-defined]
            compare_func(self.memory, other.memory),  # type: ignore[attr-defined]
            compare_func(self.gpu, other.gpu),  # type: ignore[attr-defined]
        ]
        self_storage = getattr(self, "max_storage", 99999999999999999999999)
        other_storage = getattr(other, "max_storage", 99999999999999999999999)
        results.append(compare_func(self_storage, other_storage))
        return all(results)

    def __ge__(self, other: ResourcesProtocol) -> bool:
        return self.__compare(other, lambda x, y: x >= y)

    def __gt__(self, other: ResourcesProtocol) -> bool:
        return self.__compare(other, lambda x, y: x > y)

    def __lt__(self, other: ResourcesProtocol) -> bool:
        return self.__compare(other, lambda x, y: x < y)

    def __le__(self, other: ResourcesProtocol) -> bool:
        return self.__compare(other, lambda x, y: x <= y)


@dataclass(frozen=True, eq=True, kw_only=True)
class NodeAffinity:
    """Used to set the node affinity when scheduling sessions."""

    key: str
    required_during_scheduling: bool = False

    @classmethod
    def from_dict(cls, data: dict) -> NodeAffinity:
        """Create a node affinity from a dictionary."""
        return cls(**data)


@dataclass(frozen=True, eq=True, kw_only=True)
class ResourceClass(ResourcesCompareMixin):
    """Resource class model."""

    name: str
    cpu: float
    memory: int
    max_storage: int
    gpu: int
    id: Optional[int] = None
    default: bool = False
    default_storage: int = 1
    matching: Optional[bool] = None
    node_affinities: list[NodeAffinity] = field(default_factory=list)
    tolerations: list[str] = field(default_factory=list)
    quota: str | None = None

    def __post_init__(self) -> None:
        if len(self.name) > 40:
            raise ValidationError(message="'name' cannot be longer than 40 characters.")
        if self.default_storage > self.max_storage:
            raise ValidationError(message="The default storage cannot be larger than the max allowable storage.")
        # We need to sort node affinities and tolerations to make '__eq__' reliable
        object.__setattr__(
            self, "node_affinities", sorted(self.node_affinities, key=lambda x: (x.key, x.required_during_scheduling))
        )
        object.__setattr__(self, "tolerations", sorted(self.tolerations))

    @classmethod
    def from_dict(cls, data: dict) -> ResourceClass:
        """Create the model from a plain dictionary."""
        node_affinities: list[NodeAffinity] = []
        tolerations: list[str] = []
        quota: str | None = None
        if data.get("node_affinities"):
            node_affinities = [
                NodeAffinity.from_dict(affinity) if isinstance(affinity, dict) else affinity
                for affinity in data.get("node_affinities", [])
            ]
        if isinstance(data.get("tolerations"), list):
            tolerations = [toleration for toleration in data["tolerations"]]
        if data_quota := data.get("quota"):
            if isinstance(data_quota, str):
                quota = data_quota
            elif isinstance(data_quota, Quota):
                quota = data_quota.id
        return cls(**{**data, "tolerations": tolerations, "node_affinities": node_affinities, "quota": quota})

    def is_quota_valid(self, quota: Quota) -> bool:
        """Determine if a quota is compatible with the resource class."""
        return quota >= self

    def update(self, **kwargs: dict) -> ResourceClass:
        """Update a field of the resource class and return a new copy."""
        if not kwargs:
            return self
        return ResourceClass.from_dict({**asdict(self), **kwargs})


class GpuKind(StrEnum):
    """GPU kinds for k8s."""

    NVIDIA = "nvidia.com"
    AMD = "amd.com"


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

    cpu: float
    memory: int
    gpu: int
    gpu_kind: GpuKind = GpuKind.NVIDIA
    id: Optional[str] = None

    @classmethod
    def from_dict(cls, data: dict) -> Quota:
        """Create the model from a plain dictionary."""
        gpu_kind = GpuKind.NVIDIA
        if "gpu_kind" in data:
            gpu_kind = data["gpu_kind"] if isinstance(data["gpu_kind"], GpuKind) else GpuKind[data["gpu_kind"]]
        return cls(**{**data, "gpu_kind": gpu_kind})

    def is_resource_class_compatible(self, rc: ResourceClass) -> bool:
        """Determine if a resource class is compatible with the quota."""
        return rc <= self

    def generate_id(self) -> Quota:
        """Create a new quota with its ID set to an uuid."""
        if self.id is not None:
            return self
        return self.from_dict({**asdict(self), "id": str(uuid4())})


@dataclass(frozen=True, eq=True, kw_only=True)
class Cluster:
    """K8s Cluster settings."""

    name: str
    config_name: str

    @classmethod
    def from_dict(cls, data: dict) -> Cluster:
        """Instantiate a Cluster from the dictionary."""

        if "id" in data:
            return SavedCluster(**data)
        else:
            return UnsavedCluster(**data)


@dataclass(frozen=True, eq=True, kw_only=True)
class UnsavedCluster(Cluster):
    """Unsaved, memory-only K8s Cluster settings."""


@dataclass(frozen=True, eq=True, kw_only=True)
class SavedCluster(Cluster):
    """K8s Cluster settings from the DB."""

    id: ULID


@dataclass(frozen=True, eq=True, kw_only=True)
class ResourcePool:
    """Resource pool model."""

    name: str
    classes: list[ResourceClass]
    quota: Quota | None = None
    id: int | None = None
    idle_threshold: int | None = None
    hibernation_threshold: int | None = None
    default: bool = False
    public: bool = False
    cluster: SavedCluster | None = None

    def __post_init__(self) -> None:
        """Validate the resource pool after initialization."""
        if len(self.name) > 40:
            raise ValidationError(message="'name' cannot be longer than 40 characters.")
        if self.default and not self.public:
            raise ValidationError(message="The default resource pool has to be public.")
        if self.default and self.quota is not None:
            raise ValidationError(message="A default resource pool cannot have a quota.")
        if (self.idle_threshold and self.idle_threshold < 0) or (
            self.hibernation_threshold and self.hibernation_threshold < 0
        ):
            raise ValidationError(message="Idle threshold and hibernation threshold need to be larger than 0.")

        if self.idle_threshold == 0:
            object.__setattr__(self, "idle_threshold", None)
        if self.hibernation_threshold == 0:
            object.__setattr__(self, "hibernation_threshold", None)

        default_classes = []
        for cls in list(self.classes):
            if self.quota is not None and not self.quota.is_resource_class_compatible(cls):
                raise ValidationError(
                    message=f"The resource class with name {cls.name} is not compatible with the quota."
                )
            if cls.default:
                default_classes.append(cls)
        if len(default_classes) != 1:
            raise ValidationError(message="One default class is required in each resource pool.")

    def set_quota(self, val: Quota) -> ResourcePool:
        """Set the quota for a resource pool."""
        for cls in list(self.classes):
            if not val.is_resource_class_compatible(cls):
                raise ValidationError(
                    message=f"The resource class with name {cls.name} is not compatible with the quota."
                )
        return self.from_dict({**asdict(self), "quota": val})

    def update(self, **kwargs: Any) -> ResourcePool:
        """Determine if an update to a resource pool is valid and if valid create new updated resource pool."""
        if self.default and "default" in kwargs and not kwargs["default"]:
            raise ValidationError(message="A default resource pool cannot be made non-default.")
        return ResourcePool.from_dict({**asdict(self), **kwargs})

    @classmethod
    def from_dict(cls, data: dict) -> ResourcePool:
        """Create the model from a plain dictionary."""
        cluster: SavedCluster | None = None
        quota: Quota | None = None
        classes: list[ResourceClass] = []

        if "quota" in data and isinstance(data["quota"], dict):
            quota = Quota.from_dict(data["quota"])
        elif "quota" in data and isinstance(data["quota"], Quota):
            quota = data["quota"]

        if "classes" in data and isinstance(data["classes"], set):
            classes = [ResourceClass.from_dict(c) if isinstance(c, dict) else c for c in list(data["classes"])]
        elif "classes" in data and isinstance(data["classes"], list):
            classes = [ResourceClass.from_dict(c) if isinstance(c, dict) else c for c in data["classes"]]

        if "cluster" in data:
            match data["cluster"]:
                case dict():
                    cluster = SavedCluster(**data["cluster"])
                case SavedCluster():
                    cluster = data["cluster"]
                case None:
                    cluster = None
                case unknown:
                    raise errors.ValidationError(message=f"Got unexpected cluster data {unknown} when creating model")

        return cls(
            name=data["name"],
            id=data.get("id"),
            classes=classes,
            quota=quota,
            default=data.get("default", False),
            public=data.get("public", False),
            idle_threshold=data.get("idle_threshold"),
            hibernation_threshold=data.get("hibernation_threshold"),
            cluster=cluster,
        )

    def get_resource_class(self, resource_class_id: int) -> ResourceClass | None:
        """Find a specific resource class in the resource pool by the resource class id."""
        for rc in self.classes:
            if rc.id == resource_class_id:
                return rc
        return None

    def get_default_resource_class(self) -> ResourceClass | None:
        """Find the default resource class in the pool."""
        for rc in self.classes:
            if rc.default:
                return rc
        return None

1	"""Domain models for the application."""
2
3	from __future__ import annotations	2✔
4
5	from collections.abc import Callable	2✔
6	from dataclasses import asdict, dataclass, field	2✔
7	from enum import StrEnum	2✔
8	from typing import Any, Optional, Protocol	2✔
9	from uuid import uuid4	2✔
10
11	from ulid import ULID	2✔
12
13	from renku_data_services import errors	2✔
14	from renku_data_services.errors import ValidationError	2✔
15
16
17	class ResourcesProtocol(Protocol):	2✔
18	"""Used to represent resource values present in a resource class or quota."""
19
20	@property	2✔
21	def cpu(self) -> float:	2✔
22	"""Cpu in fractional cores."""
23	...	×
24
25	@property	2✔
26	def gpu(self) -> int:	2✔
27	"""Number of GPUs."""
28	...	×
29
30	@property	2✔
31	def memory(self) -> int:	2✔
32	"""Memory in gigabytes."""
33	...	×
34
35	@property	2✔
36	def max_storage(self) -> Optional[int]:	2✔
37	"""Maximum allowable storage in gigabytes."""
38	...	×
39
40
41	class ResourcesCompareMixin:	2✔
42	"""A mixin that adds comparison operator support on ResourceClasses and Quotas."""
43
44	def __compare(	2✔
45	self,
46	other: ResourcesProtocol,
47	compare_func: Callable[[int \| float, int \| float], bool],
48	) -> bool:
49	results = [	2✔
50	compare_func(self.cpu, other.cpu), # type: ignore[attr-defined]
51	compare_func(self.memory, other.memory), # type: ignore[attr-defined]
52	compare_func(self.gpu, other.gpu), # type: ignore[attr-defined]
53	]
54	self_storage = getattr(self, "max_storage", 99999999999999999999999)	2✔
55	other_storage = getattr(other, "max_storage", 99999999999999999999999)	2✔
56	results.append(compare_func(self_storage, other_storage))	2✔
57	return all(results)	2✔
58
59	def __ge__(self, other: ResourcesProtocol) -> bool:	2✔
60	return self.__compare(other, lambda x, y: x >= y)	×
61
62	def __gt__(self, other: ResourcesProtocol) -> bool:	2✔
63	return self.__compare(other, lambda x, y: x > y)	×
64
65	def __lt__(self, other: ResourcesProtocol) -> bool:	2✔
66	return self.__compare(other, lambda x, y: x < y)	×
67
68	def __le__(self, other: ResourcesProtocol) -> bool:	2✔
69	return self.__compare(other, lambda x, y: x <= y)	2✔
70
71
72	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
73	class NodeAffinity:	2✔
74	"""Used to set the node affinity when scheduling sessions."""
75
76	key: str	2✔
77	required_during_scheduling: bool = False	2✔
78
79	@classmethod	2✔
80	def from_dict(cls, data: dict) -> NodeAffinity:	2✔
81	"""Create a node affinity from a dictionary."""
82	return cls(**data)	2✔
83
84
85	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
86	class ResourceClass(ResourcesCompareMixin):	2✔
87	"""Resource class model."""
88
89	name: str	2✔
90	cpu: float	2✔
91	memory: int	2✔
92	max_storage: int	2✔
93	gpu: int	2✔
94	id: Optional[int] = None	2✔
95	default: bool = False	2✔
96	default_storage: int = 1	2✔
97	matching: Optional[bool] = None	2✔
98	node_affinities: list[NodeAffinity] = field(default_factory=list)	2✔
99	tolerations: list[str] = field(default_factory=list)	2✔
100	quota: str \| None = None	2✔
101
102	def __post_init__(self) -> None:	2✔
103	if len(self.name) > 40:	2✔
104	raise ValidationError(message="'name' cannot be longer than 40 characters.")	1✔
105	if self.default_storage > self.max_storage:	2✔
UNCOV 106	raise ValidationError(message="The default storage cannot be larger than the max allowable storage.")	×
107	# We need to sort node affinities and tolerations to make '__eq__' reliable
108	object.__setattr__(	2✔
109	self, "node_affinities", sorted(self.node_affinities, key=lambda x: (x.key, x.required_during_scheduling))
110	)
111	object.__setattr__(self, "tolerations", sorted(self.tolerations))	2✔
112
113	@classmethod	2✔
114	def from_dict(cls, data: dict) -> ResourceClass:	2✔
115	"""Create the model from a plain dictionary."""
116	node_affinities: list[NodeAffinity] = []	2✔
117	tolerations: list[str] = []	2✔
118	quota: str \| None = None	2✔
119	if data.get("node_affinities"):	2✔
120	node_affinities = [	2✔
121	NodeAffinity.from_dict(affinity) if isinstance(affinity, dict) else affinity
122	for affinity in data.get("node_affinities", [])
123	]
124	if isinstance(data.get("tolerations"), list):	2✔
125	tolerations = [toleration for toleration in data["tolerations"]]	2✔
126	if data_quota := data.get("quota"):	2✔
127	if isinstance(data_quota, str):	1✔
128	quota = data_quota	1✔
129	elif isinstance(data_quota, Quota):	×
130	quota = data_quota.id	×
131	return cls({data, "tolerations": tolerations, "node_affinities": node_affinities, "quota": quota})	2✔
132
133	def is_quota_valid(self, quota: Quota) -> bool:	2✔
134	"""Determine if a quota is compatible with the resource class."""
135	return quota >= self	×
136
137	def update(self, **kwargs: dict) -> ResourceClass:	2✔
138	"""Update a field of the resource class and return a new copy."""
139	if not kwargs:	1✔
140	return self	×
141	return ResourceClass.from_dict({asdict(self), kwargs})	1✔
142
143
144	class GpuKind(StrEnum):	2✔
145	"""GPU kinds for k8s."""
146
147	NVIDIA = "nvidia.com"	2✔
148	AMD = "amd.com"	2✔
149
150
151	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
152	class Quota(ResourcesCompareMixin):	2✔
153	"""Quota model."""
154
155	cpu: float	2✔
156	memory: int	2✔
157	gpu: int	2✔
158	gpu_kind: GpuKind = GpuKind.NVIDIA	2✔
159	id: Optional[str] = None	2✔
160
161	@classmethod	2✔
162	def from_dict(cls, data: dict) -> Quota:	2✔
163	"""Create the model from a plain dictionary."""
164	gpu_kind = GpuKind.NVIDIA	2✔
165	if "gpu_kind" in data:	2✔
166	gpu_kind = data["gpu_kind"] if isinstance(data["gpu_kind"], GpuKind) else GpuKind[data["gpu_kind"]]	1✔
167	return cls({data, "gpu_kind": gpu_kind})	2✔
168
169	def is_resource_class_compatible(self, rc: ResourceClass) -> bool:	2✔
170	"""Determine if a resource class is compatible with the quota."""
171	return rc <= self	2✔
172
173	def generate_id(self) -> Quota:	2✔
174	"""Create a new quota with its ID set to an uuid."""
175	if self.id is not None:	1✔
176	return self	×
177	return self.from_dict({**asdict(self), "id": str(uuid4())})	1✔
178
179
180	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
181	class Cluster:	2✔
182	"""K8s Cluster settings."""
183
184	name: str	2✔
185	config_name: str	2✔
186
187	@classmethod	2✔
188	def from_dict(cls, data: dict) -> Cluster:	2✔
189	"""Instantiate a Cluster from the dictionary."""
190
191	if "id" in data:	×
192	return SavedCluster(**data)	×
193	else:
194	return UnsavedCluster(**data)	×
195
196
197	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
198	class UnsavedCluster(Cluster):	2✔
199	"""Unsaved, memory-only K8s Cluster settings."""
200
201
202	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
203	class SavedCluster(Cluster):	2✔
204	"""K8s Cluster settings from the DB."""
205
206	id: ULID	2✔
207
208
209	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
210	class ResourcePool:	2✔
211	"""Resource pool model."""
212
213	name: str	2✔
214	classes: list[ResourceClass]	2✔
215	quota: Quota \| None = None	2✔
216	id: int \| None = None	2✔
217	idle_threshold: int \| None = None	2✔
218	hibernation_threshold: int \| None = None	2✔
219	default: bool = False	2✔
220	public: bool = False	2✔
221	cluster: SavedCluster \| None = None	2✔
222
223	def __post_init__(self) -> None:	2✔
224	"""Validate the resource pool after initialization."""
225	if len(self.name) > 40:	2✔
UNCOV 226	raise ValidationError(message="'name' cannot be longer than 40 characters.")	×
227	if self.default and not self.public:	2✔
228	raise ValidationError(message="The default resource pool has to be public.")	×
229	if self.default and self.quota is not None:	2✔
230	raise ValidationError(message="A default resource pool cannot have a quota.")	×
231	if (self.idle_threshold and self.idle_threshold < 0) or (	2✔
232	self.hibernation_threshold and self.hibernation_threshold < 0
233	):
234	raise ValidationError(message="Idle threshold and hibernation threshold need to be larger than 0.")	×
235
236	if self.idle_threshold == 0:	2✔
237	object.__setattr__(self, "idle_threshold", None)	×
238	if self.hibernation_threshold == 0:	2✔
239	object.__setattr__(self, "hibernation_threshold", None)	×
240
241	default_classes = []	2✔
242	for cls in list(self.classes):	2✔
243	if self.quota is not None and not self.quota.is_resource_class_compatible(cls):	2✔
244	raise ValidationError(	1✔
245	message=f"The resource class with name {cls.name} is not compatible with the quota."
246	)
247	if cls.default:	2✔
248	default_classes.append(cls)	2✔
249	if len(default_classes) != 1:	2✔
250	raise ValidationError(message="One default class is required in each resource pool.")	1✔
251
252	def set_quota(self, val: Quota) -> ResourcePool:	2✔
253	"""Set the quota for a resource pool."""
254	for cls in list(self.classes):	1✔
255	if not val.is_resource_class_compatible(cls):	1✔
256	raise ValidationError(	×
257	message=f"The resource class with name {cls.name} is not compatible with the quota."
258	)
259	return self.from_dict({**asdict(self), "quota": val})	1✔
260
261	def update(self, **kwargs: Any) -> ResourcePool:	2✔
262	"""Determine if an update to a resource pool is valid and if valid create new updated resource pool."""
263	if self.default and "default" in kwargs and not kwargs["default"]:	1✔
264	raise ValidationError(message="A default resource pool cannot be made non-default.")	×
265	return ResourcePool.from_dict({asdict(self), kwargs})	1✔
266
267	@classmethod	2✔
268	def from_dict(cls, data: dict) -> ResourcePool:	2✔
269	"""Create the model from a plain dictionary."""
270	cluster: SavedCluster \| None = None	2✔
271	quota: Quota \| None = None	2✔
272	classes: list[ResourceClass] = []	2✔
273
274	if "quota" in data and isinstance(data["quota"], dict):	2✔
275	quota = Quota.from_dict(data["quota"])	2✔
276	elif "quota" in data and isinstance(data["quota"], Quota):	2✔
277	quota = data["quota"]	1✔
278
279	if "classes" in data and isinstance(data["classes"], set):	2✔
280	classes = [ResourceClass.from_dict(c) if isinstance(c, dict) else c for c in list(data["classes"])]	×
281	elif "classes" in data and isinstance(data["classes"], list):	2✔
282	classes = [ResourceClass.from_dict(c) if isinstance(c, dict) else c for c in data["classes"]]	2✔
283
284	if "cluster" in data:	2✔
285	match data["cluster"]:	1✔
286	case dict():	1✔
287	cluster = SavedCluster(**data["cluster"])	×
288	case SavedCluster():	1✔
289	cluster = data["cluster"]	×
290	case None:	1✔
291	cluster = None	1✔
292	case unknown:	×
293	raise errors.ValidationError(message=f"Got unexpected cluster data {unknown} when creating model")	×
294
295	return cls(	2✔
296	name=data["name"],
297	id=data.get("id"),
298	classes=classes,
299	quota=quota,
300	default=data.get("default", False),
301	public=data.get("public", False),
302	idle_threshold=data.get("idle_threshold"),
303	hibernation_threshold=data.get("hibernation_threshold"),
304	cluster=cluster,
305	)
306
307	def get_resource_class(self, resource_class_id: int) -> ResourceClass \| None:	2✔
308	"""Find a specific resource class in the resource pool by the resource class id."""
309	for rc in self.classes:	×
310	if rc.id == resource_class_id:	×
311	return rc	×
312	return None	×
313
314	def get_default_resource_class(self) -> ResourceClass \| None:	2✔
315	"""Find the default resource class in the pool."""
316	for rc in self.classes:	×
317	if rc.default:	×
318	return rc	×
319	return None	×

SwissDataScienceCenter / renku-data-services / 15844898279

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