18715077147

Committed 22 Oct 2025 11:45AM UTC coverage: 86.653% (-0.1%) from 86.802%

Build # 18715077147

Build Type

Pull #945

github

Committed by

web-flow

Commit Message

Merge 93c973951 into 5cc2b39de

Pull Request Pull Request #945: feat: add support for OpenBIS datasets

Run Details

70 of 139 new or added lines in 12 files covered. (50.36%)

4 existing lines in 3 files now uncovered.

22781 of 26290 relevant lines covered (86.65%)

1.52 hits per line

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

88.16

/components/renku_data_services/base_api/pagination.py

"""Classes and decorators used for paginating long responses."""

from collections.abc import Callable, Coroutine, Sequence
from functools import wraps
from math import ceil
from typing import Any, Concatenate, NamedTuple, ParamSpec, TypeVar, cast

from sanic import Request, json
from sanic.response import JSONResponse
from sqlalchemy import Select
from sqlalchemy.ext.asyncio import AsyncSession

from renku_data_services import errors


class PaginationRequest(NamedTuple):
    """Request for a paginated response."""

    page: int
    per_page: int

    def __post_init__(self) -> None:
        # NOTE: Postgres will fail if a value higher than what can fit in signed int64 is present in the query
        if self.page > 2**63 - 1:
            raise errors.ValidationError(message="Pagination parameter 'page' is too large")

    @property
    def offset(self) -> int:
        """Calculate an item offset required for pagination."""
        output = (self.page - 1) * self.per_page
        # NOTE: Postgres will fail if a value higher than what can fit in signed int64 is present in the query
        if output > 2**63 - 1:
            raise errors.ValidationError(
                message="Calculated pagination offset value is too large because "
                "the pagination parameter 'page' in the request is too large"
            )
        return output


class PaginationResponse(NamedTuple):
    """Paginated response parameters."""

    page: int
    per_page: int
    total: int
    total_pages: int

    def as_header(self) -> dict[str, str]:
        """Convert the instance into a dictionary that can be inserted into a HTTP header."""
        return {
            "page": str(self.page),
            "per-page": str(self.per_page),
            "total": str(self.total),
            "total-pages": str(self.total_pages),
        }


_P = ParamSpec("_P")


def paginate(
    f: Callable[Concatenate[Request, _P], Coroutine[Any, Any, tuple[Sequence[Any], int]]],
) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, JSONResponse]]:
    """Serializes the response to JSON and adds the required pagination headers to the response.

    The handler should return first the list of items and then the total count from the DB.
    """

    @wraps(f)
    async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwargs) -> JSONResponse:
        default_page_number = 1
        default_number_of_elements_per_page = 20
        query_args: dict[str, str] = request.get_args() or {}
        page_parameter = cast(int | str, query_args.get("page", default_page_number))
        try:
            page = int(page_parameter)
        except ValueError as err:
            raise errors.ValidationError(message=f"Invalid value for parameter 'page': {page_parameter}") from err
        if page < 1:
            raise errors.ValidationError(message="Parameter 'page' must be a natural number")

        per_page_parameter = cast(int | str, query_args.get("per_page", default_number_of_elements_per_page))
        try:
            per_page = int(per_page_parameter)
        except ValueError as err:
            raise errors.ValidationError(
                message=f"Invalid value for parameter 'per_page': {per_page_parameter}"
            ) from err
        if per_page < 1 or per_page > 100:
            raise errors.ValidationError(message="Parameter 'per_page' must be between 1 and 100")

        pagination_req = PaginationRequest(page, per_page)
        kwargs["pagination"] = pagination_req
        items, db_count = await f(request, *args, **kwargs)
        total_pages = ceil(db_count / per_page)

        pagination = PaginationResponse(page, per_page, db_count, total_pages)
        return json(items, headers=pagination.as_header())

    return decorated_function


_T = TypeVar("_T")


async def paginate_queries(
    req: PaginationRequest, session: AsyncSession, stmts: list[tuple[Select[tuple[_T]], int]]
) -> list[_T]:
    """Paginate several different queries as if they were part of a single table."""
    # NOTE: We ignore the possibility that a count for a statement is not accurate. I.e. the count
    # says that the statement should return 10 items but the statement truly returns 8 or vice-versa.
    # To fully account for edge cases of inaccuracry in the expected number of results
    # we would have to run every query passed in - even though the offset is so high that we would only need
    # to run 1 or 2 queries out of a large list.
    output: list[_T] = []
    max_offset = 0
    stmt_offset = 0
    offset_discount = 0
    for stmt, stmt_cnt in stmts:
        max_offset += stmt_cnt
        if req.offset >= max_offset:
            offset_discount += stmt_cnt
            continue
        stmt_offset = req.offset - offset_discount if req.offset > 0 else 0
        res_scalar = await session.scalars(stmt.offset(stmt_offset).limit(req.per_page))
        res = res_scalar.all()
        num_required = req.per_page - len(output)
        if num_required >= len(res):
            output.extend(res)
        else:
            output.extend(res[:num_required])
            return output
    return output

1	"""Classes and decorators used for paginating long responses."""
2
3	from collections.abc import Callable, Coroutine, Sequence	2✔
4	from functools import wraps	2✔
5	from math import ceil	2✔
6	from typing import Any, Concatenate, NamedTuple, ParamSpec, TypeVar, cast	2✔
7
8	from sanic import Request, json	2✔
9	from sanic.response import JSONResponse	2✔
10	from sqlalchemy import Select	2✔
11	from sqlalchemy.ext.asyncio import AsyncSession	2✔
12
13	from renku_data_services import errors	2✔
14
15
16	class PaginationRequest(NamedTuple):	2✔
17	"""Request for a paginated response."""
18
19	page: int	2✔
20	per_page: int	2✔
21
22	def __post_init__(self) -> None:	2✔
23	# NOTE: Postgres will fail if a value higher than what can fit in signed int64 is present in the query
24	if self.page > 2**63 - 1:	×
25	raise errors.ValidationError(message="Pagination parameter 'page' is too large")	×
26
27	@property	2✔
28	def offset(self) -> int:	2✔
29	"""Calculate an item offset required for pagination."""
30	output = (self.page - 1) * self.per_page	2✔
31	# NOTE: Postgres will fail if a value higher than what can fit in signed int64 is present in the query
32	if output > 2**63 - 1:	2✔
UNCOV 33	raise errors.ValidationError(	×
34	message="Calculated pagination offset value is too large because "
35	"the pagination parameter 'page' in the request is too large"
36	)
37	return output	2✔
38
39
40	class PaginationResponse(NamedTuple):	2✔
41	"""Paginated response parameters."""
42
43	page: int	2✔
44	per_page: int	2✔
45	total: int	2✔
46	total_pages: int	2✔
47
48	def as_header(self) -> dict[str, str]:	2✔
49	"""Convert the instance into a dictionary that can be inserted into a HTTP header."""
50	return {	2✔
51	"page": str(self.page),
52	"per-page": str(self.per_page),
53	"total": str(self.total),
54	"total-pages": str(self.total_pages),
55	}
56
57
58	_P = ParamSpec("_P")	2✔
59
60
61	def paginate(	2✔
62	f: Callable[Concatenate[Request, _P], Coroutine[Any, Any, tuple[Sequence[Any], int]]],
63	) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, JSONResponse]]:
64	"""Serializes the response to JSON and adds the required pagination headers to the response.
65
66	The handler should return first the list of items and then the total count from the DB.
67	"""
68
69	@wraps(f)	2✔
70	async def decorated_function(request: Request, args: _P.args, *kwargs: _P.kwargs) -> JSONResponse:	2✔
71	default_page_number = 1	2✔
72	default_number_of_elements_per_page = 20	2✔
73	query_args: dict[str, str] = request.get_args() or {}	2✔
74	page_parameter = cast(int \| str, query_args.get("page", default_page_number))	2✔
75	try:	2✔
76	page = int(page_parameter)	2✔
77	except ValueError as err:	×
78	raise errors.ValidationError(message=f"Invalid value for parameter 'page': {page_parameter}") from err	×
79	if page < 1:	2✔
80	raise errors.ValidationError(message="Parameter 'page' must be a natural number")	×
81
82	per_page_parameter = cast(int \| str, query_args.get("per_page", default_number_of_elements_per_page))	2✔
83	try:	2✔
84	per_page = int(per_page_parameter)	2✔
85	except ValueError as err:	×
86	raise errors.ValidationError(	×
87	message=f"Invalid value for parameter 'per_page': {per_page_parameter}"
88	) from err
89	if per_page < 1 or per_page > 100:	2✔
90	raise errors.ValidationError(message="Parameter 'per_page' must be between 1 and 100")	×
91
92	pagination_req = PaginationRequest(page, per_page)	2✔
93	kwargs["pagination"] = pagination_req	2✔
94	items, db_count = await f(request, args, *kwargs)	2✔
95	total_pages = ceil(db_count / per_page)	2✔
96
97	pagination = PaginationResponse(page, per_page, db_count, total_pages)	2✔
98	return json(items, headers=pagination.as_header())	2✔
99
100	return decorated_function	2✔
101
102
103	_T = TypeVar("_T")	2✔
104
105
106	async def paginate_queries(	2✔
107	req: PaginationRequest, session: AsyncSession, stmts: list[tuple[Select[tuple[_T]], int]]
108	) -> list[_T]:
109	"""Paginate several different queries as if they were part of a single table."""
110	# NOTE: We ignore the possibility that a count for a statement is not accurate. I.e. the count
111	# says that the statement should return 10 items but the statement truly returns 8 or vice-versa.
112	# To fully account for edge cases of inaccuracry in the expected number of results
113	# we would have to run every query passed in - even though the offset is so high that we would only need
114	# to run 1 or 2 queries out of a large list.
115	output: list[_T] = []	2✔
116	max_offset = 0	2✔
117	stmt_offset = 0	2✔
118	offset_discount = 0	2✔
119	for stmt, stmt_cnt in stmts:	2✔
120	max_offset += stmt_cnt	2✔
121	if req.offset >= max_offset:	2✔
122	offset_discount += stmt_cnt	2✔
123	continue	2✔
124	stmt_offset = req.offset - offset_discount if req.offset > 0 else 0	2✔
125	res_scalar = await session.scalars(stmt.offset(stmt_offset).limit(req.per_page))	2✔
126	res = res_scalar.all()	2✔
127	num_required = req.per_page - len(output)	2✔
128	if num_required >= len(res):	2✔
129	output.extend(res)	2✔
130	else:
131	output.extend(res[:num_required])	1✔
132	return output	1✔
133	return output	2✔

SwissDataScienceCenter / renku-data-services / 18715077147

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