22672253664

Committed 04 Mar 2026 01:37PM UTC coverage: 87.187% (-0.004%) from 87.191%

Build # 22672253664

Build Type

push

github

Committed by

web-flow

Commit Message

feat: Enable resource requests tracking (#1207)

* Record certain data in a `resource_request_log table`
* Snapshots are stored periodically from data_tasks
* Resource request tracking is disabled by default, requires `ENABLE_RESOURCE_REQUEST_TRACKING=true` to be in the env

Coverage Stats

841 of 970 new or added lines in 16 files covered. (86.7%)

4 existing lines in 2 files now uncovered.

25006 of 28681 relevant lines covered (87.19%)

1.52 hits per line

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

92.42

/components/renku_data_services/utils/sqlalchemy.py

"""Utilities for SQLAlchemy."""

from pathlib import PurePosixPath
from typing import cast

import asyncpg
from sqlalchemy import Dialect, types
from ulid import ULID

from renku_data_services.resource_usage.model import ComputeCapacity, Credit, DataSize


def get_postgres_error_code(exception: Exception) -> str | None:
    """Traverse exceptions to find and return the SQLSTATE error code.

    sqlalchemy hides the underlying database error in multiple layers
    of exceptions, making it not so straight forward to reach. This
    method is a helper to retrieve the postgresql error code which
    exactly defines the error situation.
    """
    exc: BaseException | None = exception
    while exc:
        if isinstance(exc, asyncpg.PostgresError):
            # impl note: this attribute is dynamic so there is no way to find it's definition
            # the documentation shows that it is available.
            return cast(str | None, exc.sqlstate)
        exc = exc.__cause__
    return None


class CreditType(types.TypeDecorator):
    """Convert Credit values to/from db."""

    impl = types.Integer
    cache_ok = True

    def process_bind_param(self, value: Credit | None, dialect: Dialect) -> int | None:
        """Transform to int."""
        return value.value if value is not None else None

    def process_result_value(self, value: int | None, dialect: Dialect) -> Credit | None:
        """Transform into Credit."""
        return Credit.from_int(value) if value is not None else None

    def process_literal_param(self, value: Credit | None, dialect: Dialect) -> str:
        """Return literal."""
        return str(value) if value is not None else ""


class ComputeCapacityType(types.TypeDecorator):
    """Convert ComputeCapacity values to/from db."""

    impl = types.Float

    def process_bind_param(self, value: ComputeCapacity | None, dialect: Dialect) -> float | None:
        """Transform value into a float."""
        if value is None:
            return None
        return value.cores

    def process_result_value(self, value: float | None, dialect: Dialect) -> ComputeCapacity | None:
        """Transform a float into ComputeCapacity value."""
        if value is None:
            return None
        return ComputeCapacity.from_cores(value)

    def process_literal_param(self, value: ComputeCapacity | None, dialect: Dialect) -> str:
        """Return a literal representation."""
        return str(value) if value is not None else ""


class DataSizeType(types.TypeDecorator):
    """Convert DataSize values to/from db."""

    impl = types.Float

    def process_bind_param(self, value: DataSize | None, dialect: Dialect) -> float | None:
        """Convert to db value."""
        return value.bytes if value is not None else None

    def process_result_value(self, value: float | None, dialect: Dialect) -> DataSize | None:
        """Convert to DataSize value."""
        return DataSize.from_bytes(value) if value is not None else None

    def process_literal_param(self, value: DataSize | None, dialect: Dialect) -> str:
        """Convert to literal."""
        return str(value) if value is not None else ""


class ULIDType(types.TypeDecorator):
    """Wrapper type for ULID <--> str conversion."""

    impl = types.String
    cache_ok = True

    def process_bind_param(self, value: ULID | None, dialect: Dialect) -> str | None:
        """Transform value for storing in the database."""
        if value is None:
            return None
        return str(value)

    def process_result_value(self, value: str | None, dialect: Dialect) -> ULID | None:
        """Transform string from database into ULID."""
        if value is None:
            return None
        return cast(ULID, ULID.from_str(value))  # cast because mypy doesn't understand ULID type annotations


class PurePosixPathType(types.TypeDecorator):
    """Wrapper type for Path <--> str conversion."""

    impl = types.String
    cache_ok = True

    def process_bind_param(self, value: PurePosixPath | str | None, dialect: Dialect) -> str | None:
        """Transform value for storing in the database."""
        if value is None:
            return None
        elif isinstance(value, str):
            return value
        else:
            return value.as_posix()

    def process_result_value(self, value: str | None, dialect: Dialect) -> PurePosixPath | None:
        """Transform string from database into PosixPath."""
        if value is None:
            return None
        return PurePosixPath(value)

1	"""Utilities for SQLAlchemy."""
2
3	from pathlib import PurePosixPath	2✔
4	from typing import cast	2✔
5
6	import asyncpg	2✔
7	from sqlalchemy import Dialect, types	2✔
8	from ulid import ULID	2✔
9
10	from renku_data_services.resource_usage.model import ComputeCapacity, Credit, DataSize	2✔
11
12
13	def get_postgres_error_code(exception: Exception) -> str \| None:	2✔
14	"""Traverse exceptions to find and return the SQLSTATE error code.
15
16	sqlalchemy hides the underlying database error in multiple layers
17	of exceptions, making it not so straight forward to reach. This
18	method is a helper to retrieve the postgresql error code which
19	exactly defines the error situation.
20	"""
21	exc: BaseException \| None = exception	1✔
22	while exc:	1✔
23	if isinstance(exc, asyncpg.PostgresError):	1✔
24	# impl note: this attribute is dynamic so there is no way to find it's definition
25	# the documentation shows that it is available.
26	return cast(str \| None, exc.sqlstate)	1✔
27	exc = exc.__cause__	1✔
NEW 28	return None	×
29
30
31	class CreditType(types.TypeDecorator):	2✔
32	"""Convert Credit values to/from db."""
33
34	impl = types.Integer	2✔
35	cache_ok = True	2✔
36
37	def process_bind_param(self, value: Credit \| None, dialect: Dialect) -> int \| None:	2✔
38	"""Transform to int."""
39	return value.value if value is not None else None	2✔
40
41	def process_result_value(self, value: int \| None, dialect: Dialect) -> Credit \| None:	2✔
42	"""Transform into Credit."""
43	return Credit.from_int(value) if value is not None else None	1✔
44
45	def process_literal_param(self, value: Credit \| None, dialect: Dialect) -> str:	2✔
46	"""Return literal."""
NEW 47	return str(value) if value is not None else ""	×
48
49
50	class ComputeCapacityType(types.TypeDecorator):	2✔
51	"""Convert ComputeCapacity values to/from db."""
52
53	impl = types.Float	2✔
54
55	def process_bind_param(self, value: ComputeCapacity \| None, dialect: Dialect) -> float \| None:	2✔
56	"""Transform value into a float."""
57	if value is None:	1✔
58	return None	1✔
59	return value.cores	1✔
60
61	def process_result_value(self, value: float \| None, dialect: Dialect) -> ComputeCapacity \| None:	2✔
62	"""Transform a float into ComputeCapacity value."""
63	if value is None:	1✔
64	return None	1✔
65	return ComputeCapacity.from_cores(value)	1✔
66
67	def process_literal_param(self, value: ComputeCapacity \| None, dialect: Dialect) -> str:	2✔
68	"""Return a literal representation."""
NEW 69	return str(value) if value is not None else ""	×
70
71
72	class DataSizeType(types.TypeDecorator):	2✔
73	"""Convert DataSize values to/from db."""
74
75	impl = types.Float	2✔
76
77	def process_bind_param(self, value: DataSize \| None, dialect: Dialect) -> float \| None:	2✔
78	"""Convert to db value."""
79	return value.bytes if value is not None else None	1✔
80
81	def process_result_value(self, value: float \| None, dialect: Dialect) -> DataSize \| None:	2✔
82	"""Convert to DataSize value."""
83	return DataSize.from_bytes(value) if value is not None else None	1✔
84
85	def process_literal_param(self, value: DataSize \| None, dialect: Dialect) -> str:	2✔
86	"""Convert to literal."""
NEW 87	return str(value) if value is not None else ""	×
88
89
90	class ULIDType(types.TypeDecorator):	2✔
91	"""Wrapper type for ULID <--> str conversion."""
92
93	impl = types.String	2✔
94	cache_ok = True	2✔
95
96	def process_bind_param(self, value: ULID \| None, dialect: Dialect) -> str \| None:	2✔
97	"""Transform value for storing in the database."""
98	if value is None:	2✔
99	return None	2✔
100	return str(value)	2✔
101
102	def process_result_value(self, value: str \| None, dialect: Dialect) -> ULID \| None:	2✔
103	"""Transform string from database into ULID."""
104	if value is None:	2✔
105	return None	2✔
106	return cast(ULID, ULID.from_str(value)) # cast because mypy doesn't understand ULID type annotations	2✔
107
108
109	class PurePosixPathType(types.TypeDecorator):	2✔
110	"""Wrapper type for Path <--> str conversion."""
111
112	impl = types.String	2✔
113	cache_ok = True	2✔
114
115	def process_bind_param(self, value: PurePosixPath \| str \| None, dialect: Dialect) -> str \| None:	2✔
116	"""Transform value for storing in the database."""
117	if value is None:	2✔
118	return None	2✔
119	elif isinstance(value, str):	2✔
120	return value	×
121	else:
122	return value.as_posix()	2✔
123
124	def process_result_value(self, value: str \| None, dialect: Dialect) -> PurePosixPath \| None:	2✔
125	"""Transform string from database into PosixPath."""
126	if value is None:	2✔
127	return None	2✔
128	return PurePosixPath(value)	2✔

SwissDataScienceCenter / renku-data-services / 22672253664

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