12273418059

Committed 11 Dec 2024 09:27AM UTC coverage: 85.74% (+0.02%) from 85.718%

Build # 12273418059

Build Type

push

github

Committed by

web-flow

Commit Message

fix: do not skip the k8s cache by default (#571)

Run Details

7 of 9 new or added lines in 1 file covered. (77.78%)

1 existing line in 1 file now uncovered.

14526 of 16942 relevant lines covered (85.74%)

1.52 hits per line

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

98.36

/components/renku_data_services/base_api/misc.py

"""Common blueprints."""

from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from functools import wraps
from typing import Any, Concatenate, NoReturn, ParamSpec, TypeVar, cast

from pydantic import BaseModel, RootModel
from sanic import Request, json
from sanic.response import JSONResponse
from sanic_ext import validate

from renku_data_services import errors
from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint


@dataclass(kw_only=True)
class MiscBP(CustomBlueprint):
    """Server contains all handlers for CRC and the configuration."""

    apispec: dict[str, Any]
    version: str

    def get_apispec(self) -> BlueprintFactoryResponse:
        """Servers the OpenAPI specification."""

        async def _get_apispec(_: Request) -> JSONResponse:
            return json(self.apispec)

        return "/spec.json", ["GET"], _get_apispec

    def get_error(self) -> BlueprintFactoryResponse:
        """Returns a sample error response."""

        async def _get_error(_: Request) -> NoReturn:
            raise errors.ValidationError(message="Sample validation error")

        return "/error", ["GET"], _get_error

    def get_version(self) -> BlueprintFactoryResponse:
        """Returns the version."""

        async def _get_version(_: Request) -> JSONResponse:
            return json({"version": self.version})

        return "/version", ["GET"], _get_version


_T = TypeVar("_T")
_P = ParamSpec("_P")


def validate_db_ids(f: Callable[_P, Awaitable[_T]]) -> Callable[_P, Coroutine[Any, Any, _T]]:
    """Decorator for a Sanic handler that errors out if passed in IDs are outside of the valid range for postgres."""

    @wraps(f)
    async def decorated_function(*args: _P.args, **kwargs: _P.kwargs) -> _T:
        resource_pool_id = cast(int | None, kwargs.get("resource_pool_id"))
        class_id = cast(int | None, kwargs.get("class_id"))
        min_val = 1  # postgres primary keys start at 1
        max_val = 2_147_483_647  # the max value for a default postgres primary key sequence
        if resource_pool_id and not min_val <= resource_pool_id <= max_val:
            raise errors.ValidationError(
                message=f"The provided resource pool ID is outside of the allowed range [{min_val}, {max_val}]"
            )
        if class_id and not min_val <= class_id <= max_val:
            raise errors.ValidationError(
                message=f"The provided resource class ID is outside of the allowed range [{min_val}, {max_val}]"
            )
        response = await f(*args, **kwargs)
        return response

    return decorated_function


def validate_query(
    query: type[BaseModel],
) -> Callable[
    [Callable[Concatenate[Request, _P], Awaitable[_T]]],
    Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]],
]:
    """Decorator for sanic query parameter validation.

    Should be removed once sanic fixes this error in their validation code.
    """

    def decorator(
        f: Callable[Concatenate[Request, _P], Awaitable[_T]],
    ) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]]:
        @wraps(f)
        async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwargs) -> _T:
            try:
                return await validate(query=query)(f)(request, *args, **kwargs)
            except KeyError:
                raise errors.ValidationError(message="Failed to validate the query parameters")

        return decorated_function

    return decorator


def validate_body_root_model(
    json: type[RootModel],
) -> Callable[
    [Callable[Concatenate[Request, _P], Awaitable[_T]]],
    Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]],
]:
    """Decorator for sanic json payload validation when the model is derived from RootModel.

    Should be removed once sanic fixes this error in their validation code.
    Issue link: https://github.com/sanic-org/sanic-ext/issues/198
    """

    def decorator(
        f: Callable[Concatenate[Request, _P], Awaitable[_T]],
    ) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]]:
        @wraps(f)
        async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwargs) -> _T:
            if request.json is not None:
                request.parsed_json = {"root": request.parsed_json}  # type: ignore[assignment]
            return await validate(json=json)(f)(request, *args, **kwargs)

        return decorated_function

    return decorator

1	"""Common blueprints."""
2
3	from collections.abc import Awaitable, Callable, Coroutine	2✔
4	from dataclasses import dataclass	2✔
5	from functools import wraps	2✔
6	from typing import Any, Concatenate, NoReturn, ParamSpec, TypeVar, cast	2✔
7
8	from pydantic import BaseModel, RootModel	2✔
9	from sanic import Request, json	2✔
10	from sanic.response import JSONResponse	2✔
11	from sanic_ext import validate	2✔
12
13	from renku_data_services import errors	2✔
14	from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint	2✔
15
16
17	@dataclass(kw_only=True)	2✔
18	class MiscBP(CustomBlueprint):	2✔
19	"""Server contains all handlers for CRC and the configuration."""
20
21	apispec: dict[str, Any]	2✔
22	version: str	2✔
23
24	def get_apispec(self) -> BlueprintFactoryResponse:	2✔
25	"""Servers the OpenAPI specification."""
26
27	async def _get_apispec(_: Request) -> JSONResponse:	2✔
28	return json(self.apispec)	1✔
29
30	return "/spec.json", ["GET"], _get_apispec	2✔
31
32	def get_error(self) -> BlueprintFactoryResponse:	2✔
33	"""Returns a sample error response."""
34
35	async def _get_error(_: Request) -> NoReturn:	2✔
36	raise errors.ValidationError(message="Sample validation error")	1✔
37
38	return "/error", ["GET"], _get_error	2✔
39
40	def get_version(self) -> BlueprintFactoryResponse:	2✔
41	"""Returns the version."""
42
43	async def _get_version(_: Request) -> JSONResponse:	2✔
44	return json({"version": self.version})	2✔
45
46	return "/version", ["GET"], _get_version	2✔
47
48
49	_T = TypeVar("_T")	2✔
50	_P = ParamSpec("_P")	2✔
51
52
53	def validate_db_ids(f: Callable[_P, Awaitable[_T]]) -> Callable[_P, Coroutine[Any, Any, _T]]:	2✔
54	"""Decorator for a Sanic handler that errors out if passed in IDs are outside of the valid range for postgres."""
55
56	@wraps(f)	2✔
57	async def decorated_function(args: _P.args, *kwargs: _P.kwargs) -> _T:	2✔
58	resource_pool_id = cast(int \| None, kwargs.get("resource_pool_id"))	2✔
59	class_id = cast(int \| None, kwargs.get("class_id"))	2✔
60	min_val = 1 # postgres primary keys start at 1	2✔
61	max_val = 2_147_483_647 # the max value for a default postgres primary key sequence	2✔
62	if resource_pool_id and not min_val <= resource_pool_id <= max_val:	2✔
63	raise errors.ValidationError(	1✔
64	message=f"The provided resource pool ID is outside of the allowed range [{min_val}, {max_val}]"
65	)
66	if class_id and not min_val <= class_id <= max_val:	2✔
UNCOV 67	raise errors.ValidationError(	×
68	message=f"The provided resource class ID is outside of the allowed range [{min_val}, {max_val}]"
69	)
70	response = await f(args, *kwargs)	2✔
71	return response	2✔
72
73	return decorated_function	2✔
74
75
76	def validate_query(	2✔
77	query: type[BaseModel],
78	) -> Callable[
79	[Callable[Concatenate[Request, _P], Awaitable[_T]]],
80	Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]],
81	]:
82	"""Decorator for sanic query parameter validation.
83
84	Should be removed once sanic fixes this error in their validation code.
85	"""
86
87	def decorator(	2✔
88	f: Callable[Concatenate[Request, _P], Awaitable[_T]],
89	) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]]:
90	@wraps(f)	2✔
91	async def decorated_function(request: Request, args: _P.args, *kwargs: _P.kwargs) -> _T:	2✔
92	try:	2✔
93	return await validate(query=query)(f)(request, args, *kwargs)	2✔
94	except KeyError:	2✔
95	raise errors.ValidationError(message="Failed to validate the query parameters")	1✔
96
97	return decorated_function	2✔
98
99	return decorator	2✔
100
101
102	def validate_body_root_model(	2✔
103	json: type[RootModel],
104	) -> Callable[
105	[Callable[Concatenate[Request, _P], Awaitable[_T]]],
106	Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]],
107	]:
108	"""Decorator for sanic json payload validation when the model is derived from RootModel.
109
110	Should be removed once sanic fixes this error in their validation code.
111	Issue link: https://github.com/sanic-org/sanic-ext/issues/198
112	"""
113
114	def decorator(	2✔
115	f: Callable[Concatenate[Request, _P], Awaitable[_T]],
116	) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]]:
117	@wraps(f)	2✔
118	async def decorated_function(request: Request, args: _P.args, *kwargs: _P.kwargs) -> _T:	2✔
119	if request.json is not None:	2✔
120	request.parsed_json = {"root": request.parsed_json} # type: ignore[assignment]	2✔
121	return await validate(json=json)(f)(request, args, *kwargs)	2✔
122
123	return decorated_function	2✔
124
125	return decorator	2✔

SwissDataScienceCenter / renku-data-services / 12273418059

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