10027921963

Committed 21 Jul 2024 11:23AM UTC coverage: 90.93% (-0.01%) from 90.941%

Build # 10027921963

Build Type

push

github

Committed by

web-flow

Commit Message

refactor routing code into `rolo.routing` package (#18)

Run Details

329 of 356 new or added lines in 13 files covered. (92.42%)

1604 of 1764 relevant lines covered (90.93%)

2.73 hits per line

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

93.18

/rolo/routing/handler.py

import json
import logging
import typing as t

from werkzeug import Response as WerkzeugResponse

try:
    import pydantic  # noqa

    ENABLE_PYDANTIC = True
except ImportError:
    ENABLE_PYDANTIC = False

from rolo.request import Request
from rolo.response import Response

from .router import Dispatcher, RequestArguments

LOG = logging.getLogger(__name__)

ResultValue = t.Union[
    WerkzeugResponse,
    str,
    bytes,
    dict[str, t.Any],  # a JSON dict
    list[t.Any],
]


class Handler(t.Protocol):
    """
    A protocol used by a ``Router`` together with the dispatcher created with ``handler_dispatcher``. Endpoints added
    with this protocol take as first argument the HTTP request object, and then as keyword arguments the request
    parameters added in the rule. This makes it work very similar to flask routes.

    Example code could look like this::

        def my_route(request: Request, organization: str, repo: str):
            return {"something": "returned as json response"}

        router = Router(dispatcher=handler_dispatcher)
        router.add("/<organization>/<repo>", endpoint=my_route)

    """

    def __call__(self, request: Request, **kwargs) -> ResultValue:
        """
        Handle the given request.

        :param request: the HTTP request object
        :param kwargs: the url request parameters
        :return: a string or bytes value, a dict to create a json response, or a raw werkzeug Response object.
        """
        raise NotImplementedError


class HandlerDispatcher:
    def __init__(self, json_encoder: t.Type[json.JSONEncoder] = None):
        self.json_encoder = json_encoder

    def __call__(
        self, request: Request, endpoint: t.Callable, request_args: RequestArguments
    ) -> Response:
        result = self.invoke_endpoint(request, endpoint, request_args)
        return self.to_response(result)

    def invoke_endpoint(
        self,
        request: Request,
        endpoint: t.Callable,
        request_args: RequestArguments,
    ) -> t.Any:
        return endpoint(request, **request_args)

    def to_response(self, value: ResultValue) -> Response:
        if isinstance(value, WerkzeugResponse):
            return value

        response = Response()
        if value is None:
            return response

        self.populate_response(response, value)
        return response

    def populate_response(self, response: Response, value: ResultValue):
        if isinstance(value, (str, bytes, bytearray)):
            response.data = value
        elif isinstance(value, (dict, list)):
            response.data = json.dumps(value, cls=self.json_encoder)
            response.mimetype = "application/json"
        else:
            raise ValueError("unhandled result type %s", type(value))


def handler_dispatcher(json_encoder: t.Type[json.JSONEncoder] = None) -> Dispatcher[Handler]:
    """
    Creates a Dispatcher that treats endpoints like callables of the ``Handler`` Protocol.

    :param json_encoder: optionally the json encoder class to use for translating responses
    :return: a new dispatcher
    """
    if ENABLE_PYDANTIC:
        from rolo.routing.pydantic import PydanticHandlerDispatcher

        return PydanticHandlerDispatcher(json_encoder)

    return HandlerDispatcher(json_encoder)

1	import json	3✔
2	import logging	3✔
3	import typing as t	3✔
4
5	from werkzeug import Response as WerkzeugResponse	3✔
6
7	try:	3✔
8	import pydantic # noqa	3✔
9
10	ENABLE_PYDANTIC = True	3✔
NEW 11	except ImportError:	×
NEW 12	ENABLE_PYDANTIC = False	×
13
14	from rolo.request import Request	3✔
15	from rolo.response import Response	3✔
16
17	from .router import Dispatcher, RequestArguments	3✔
18
19	LOG = logging.getLogger(__name__)	3✔
20
21	ResultValue = t.Union[	3✔
22	WerkzeugResponse,
23	str,
24	bytes,
25	dict[str, t.Any], # a JSON dict
26	list[t.Any],
27	]
28
29
30	class Handler(t.Protocol):	3✔
31	"""
32	A protocol used by a ``Router`` together with the dispatcher created with ``handler_dispatcher``. Endpoints added
33	with this protocol take as first argument the HTTP request object, and then as keyword arguments the request
34	parameters added in the rule. This makes it work very similar to flask routes.
35
36	Example code could look like this::
37
38	def my_route(request: Request, organization: str, repo: str):
39	return {"something": "returned as json response"}
40
41	router = Router(dispatcher=handler_dispatcher)
42	router.add("/<organization>/<repo>", endpoint=my_route)
43
44	"""
45
46	def __call__(self, request: Request, **kwargs) -> ResultValue:	3✔
47	"""
48	Handle the given request.
49
50	:param request: the HTTP request object
51	:param kwargs: the url request parameters
52	:return: a string or bytes value, a dict to create a json response, or a raw werkzeug Response object.
53	"""
54	raise NotImplementedError
55
56
57	class HandlerDispatcher:	3✔
58	def __init__(self, json_encoder: t.Type[json.JSONEncoder] = None):	3✔
59	self.json_encoder = json_encoder	3✔
60
61	def __call__(	3✔
62	self, request: Request, endpoint: t.Callable, request_args: RequestArguments
63	) -> Response:
64	result = self.invoke_endpoint(request, endpoint, request_args)	3✔
65	return self.to_response(result)	3✔
66
67	def invoke_endpoint(	3✔
68	self,
69	request: Request,
70	endpoint: t.Callable,
71	request_args: RequestArguments,
72	) -> t.Any:
73	return endpoint(request, **request_args)	3✔
74
75	def to_response(self, value: ResultValue) -> Response:	3✔
76	if isinstance(value, WerkzeugResponse):	3✔
77	return value	3✔
78
79	response = Response()	3✔
80	if value is None:	3✔
81	return response	3✔
82
83	self.populate_response(response, value)	3✔
84	return response	3✔
85
86	def populate_response(self, response: Response, value: ResultValue):	3✔
87	if isinstance(value, (str, bytes, bytearray)):	3✔
88	response.data = value	3✔
89	elif isinstance(value, (dict, list)):	3✔
90	response.data = json.dumps(value, cls=self.json_encoder)	3✔
91	response.mimetype = "application/json"	3✔
92	else:
NEW 93	raise ValueError("unhandled result type %s", type(value))	×
94
95
96	def handler_dispatcher(json_encoder: t.Type[json.JSONEncoder] = None) -> Dispatcher[Handler]:	3✔
97	"""
98	Creates a Dispatcher that treats endpoints like callables of the ``Handler`` Protocol.
99
100	:param json_encoder: optionally the json encoder class to use for translating responses
101	:return: a new dispatcher
102	"""
103	if ENABLE_PYDANTIC:	3✔
104	from rolo.routing.pydantic import PydanticHandlerDispatcher	3✔
105
106	return PydanticHandlerDispatcher(json_encoder)	3✔
107
108	return HandlerDispatcher(json_encoder)	3✔

localstack / rolo / 10027921963

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