PK2O?MY!!dataclassesapi/__init__.py""" ASGI App using dataclasses module for request/response objects """ __version__ = '0.2.0' from dataclassesapi.app import asgi_app from dataclassesapi.method import MethodType from dataclassesapi.router import Route __all__ = [asgi_app.__name__, MethodType.__name__, Route.__name__] PK 2O+dͅ dataclassesapi/app.pyfrom http import HTTPStatus from logging import getLogger from typing import Any, Awaitable, Callable, Coroutine, Dict, Iterable from urllib import parse import orjson from dataclassesjson.exceptions import DeserializationError from dataclassesapi.exceptions import MethodNotFoundError, PathNotFoundError from dataclassesapi.request import as_request from dataclassesapi.response import AsgiResponse, Response from dataclassesapi.response import as_asgi as as_asgi_response from dataclassesapi.router import Route, route from dataclassesapi.router import router as http_router logger = getLogger(__name__) Scope = Dict[str, Any] Receiver = Callable[[], Awaitable[Dict[str, Any]]] Sender = Callable[[Dict[str, Any]], Awaitable[None]] AsgiCallable = Callable[[Scope, Receiver, Sender], Coroutine[Any, Any, None]] def asgi_app(routes: Iterable[Route]) -> AsgiCallable: router = http_router(routes) async def handler(scope: Scope, receive: Receiver, send: Sender) -> None: headers = scope['headers'] try: resolved = route(router, scope['path'], scope['method']) except PathNotFoundError: await _send_asgi_response(send, AsgiResponse(HTTPStatus.NOT_FOUND)) except MethodNotFoundError: await _send_asgi_response( send, AsgiResponse(HTTPStatus.METHOD_NOT_ALLOWED) ) else: query_dict = _get_query_dict(scope) body = await _read_body(receive) request_cls = resolved.route.caller.__annotations__[ # type: ignore 'req' ] try: request = as_request( request_cls=request_cls, path_args=resolved.path_args, query_dict=query_dict, headers=headers, body=body, ) except DeserializationError as error: logger.warning(f"DeserializationError {error.message}") asgi_response = AsgiResponse( status_code=HTTPStatus.BAD_REQUEST, body=orjson.dumps(error.dict), headers=headers, ) await _send_asgi_response(send, asgi_response) else: response = resolved.route.caller(request) # type: ignore await _send_response(send, response) return handler def _get_query_dict(scope: Dict[str, Any]) -> Dict[str, Any]: qs = parse.parse_qs(scope['query_string'].decode()) return qs async def _read_body(receive: Callable[[], Awaitable[Dict[str, Any]]]) -> Any: body = b'' more_body = True while more_body: message = await receive() body += message.get('body', b'') more_body = message.get('more_body', False) return body async def _send_response( send: Callable[[Dict[str, Any]], Awaitable[None]], response: Response ) -> None: asgi_response = as_asgi_response(response) await _send_asgi_response(send, asgi_response) async def _send_asgi_response( send: Callable[[Dict[str, Any]], Awaitable[None]], asgi_response: AsgiResponse, ) -> None: await send( { 'type': 'http.response.start', 'status': asgi_response.status_code.value, 'headers': asgi_response.headers, } ) await send({'type': 'http.response.body', 'body': asgi_response.body}) PK1O*ؿ#dataclassesapi/exceptions.pyclass DataClassesApi(Exception): ... class MethodNotFoundError(DataClassesApi): ... class PathNotFoundError(DataClassesApi): ... PKS 1Oc:#dataclassesapi/method.pyfrom enum import Enum class MethodType(Enum): GET = 'GET' POST = 'POST' PATCH = 'PATCH' PUT = 'PUT' DELETE = 'DELETE' OPTIONS = 'OPTIONS' HEAD = 'HEAD' TRACE = 'TRACE' CONNECT = 'CONNECT' PK1O`dataclassesapi/request.pyfrom collections.abc import Iterable from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple, Type import orjson from dataclassesjson import asdataclass @dataclass class PathArgs: ... @dataclass class Query: ... @dataclass class Headers: ... @dataclass class Body: ... @dataclass class Request: path_args: Optional[PathArgs] = None query: Optional[Query] = None headers: Optional[Headers] = None body: Optional[Body] = None AsgiPathArgs = Dict[str, Any] AsgiQueryDict = Dict[str, List[str]] AsgiHeaders = List[Tuple[bytes, bytes]] def as_request( request_cls: Type[Request], path_args: AsgiPathArgs = {}, query_dict: AsgiQueryDict = {}, headers: AsgiHeaders = [], body: bytes = b'', ) -> Request: annotations = getattr(request_cls, '__annotations__', {}) path_args_cls = annotations.get('path_args', PathArgs) query_cls = annotations.get('query', Query) headers_cls = annotations.get('headers', Headers) body_cls = annotations.get('body', Body) return request_cls( path_args=asdataclass(path_args, path_args_cls), query=get_query(query_cls, query_dict), headers=get_headers(headers_cls, headers), body=asdataclass(orjson.loads(body) if body else {}, body_cls), ) def get_query(cls: Type[Query], query_dict: AsgiQueryDict) -> Query: jsondict: Dict[str, Any] = {} annotations = getattr(cls, '__annotations__', {}) for key in annotations.keys(): values = query_dict.get(key) if values: jsondict[key] = ( values[0] if isinstance(values, Iterable) else values ) else: jsondict[key] = values return asdataclass(jsondict, cls) # type: ignore def get_headers(cls: Type[Headers], headers: AsgiHeaders) -> Headers: jsondict = {} annotations = getattr(cls, '__annotations__', {}) for key in annotations.keys(): for key_h, value in headers: if key == key_h.decode().lower().replace('-', '_'): break else: continue jsondict[key] = value.decode() return asdataclass(jsondict, cls) # type: ignore PKO1O Tdataclassesapi/response.pyfrom dataclasses import dataclass, field from http import HTTPStatus from typing import Optional from dataclassesjson import asjson from dataclassesapi.request import AsgiHeaders @dataclass class Headers: ... @dataclass class Body: ... @dataclass class Response: status_code: HTTPStatus headers: Optional[Headers] = None body: Optional[Body] = None @dataclass class AsgiResponse: status_code: HTTPStatus headers: AsgiHeaders = field(default_factory=list) body: bytes = b'' def as_asgi(response: Response) -> AsgiResponse: headers = as_asgi_headers(response.headers) if response.headers else [] body = asjson(response.body) if response.body else b'' if body: headers.append((b'content-type', b'application/json')) headers.append((b'content-length', str(len(body)).encode())) return AsgiResponse( status_code=response.status_code, headers=headers, body=body ) def as_asgi_headers(headers: Headers) -> AsgiHeaders: return [ ( field.replace('_', '-').encode(), str(getattr(headers, field)).encode(), ) for field in type(headers).__annotations__.keys() ] PK2OB= dataclassesapi/router.pyimport re from dataclasses import dataclass from typing import ( Any, Callable, DefaultDict, Dict, Iterable, Optional, Pattern, ) from dataclassesapi.exceptions import MethodNotFoundError, PathNotFoundError from dataclassesapi.method import MethodType from dataclassesapi.request import Request from dataclassesapi.response import Response @dataclass class RoutesTreeRegex: name: str compiled_re: Optional[Pattern[Any]] class RoutesTree(DefaultDict[str, Any]): regex: Optional[RoutesTreeRegex] = None def __init__(self) -> None: super().__init__(RoutesTree) @dataclass class Route: path_pattern: str method: MethodType caller: Callable[[Request], Response] @dataclass class ResolvedRoute: route: Route path_args: Dict[str, Any] path: str def router(routes: Iterable[Route]) -> RoutesTree: routes_tree = RoutesTree() path_regex = re.compile(r'\{(?P[^/:]+)(:(?P[^/:]+))?\}') for route in routes: path_pattern_parts = _split_path(route.path_pattern) routes_tree_tmp = routes_tree for path_pattern_part in path_pattern_parts: match = path_regex.match(path_pattern_part) if match: group = match.groupdict() pattern = group.get('pattern') regex: Optional[Pattern[Any]] = re.compile( pattern ) if pattern else None routes_tree_tmp.regex = RoutesTreeRegex( group['name'], compiled_re=regex ) routes_tree_tmp = routes_tree_tmp[routes_tree_tmp.regex.name] continue routes_tree_tmp = routes_tree_tmp[path_pattern_part] routes_tree_tmp[route.method.value] = route return routes_tree def route(routes_tree: RoutesTree, path: str, method: str) -> ResolvedRoute: path_parts = _split_path(path) path_args = {} for path_part in path_parts: if path_part in routes_tree: routes_tree = routes_tree[path_part] continue if routes_tree.regex: compiled_re = routes_tree.regex.compiled_re match = compiled_re.match(path_part) if compiled_re else None if not match and compiled_re: raise PathNotFoundError(path) path_args[routes_tree.regex.name] = path_part routes_tree = routes_tree[routes_tree.regex.name] continue raise PathNotFoundError(path) if method not in routes_tree: raise MethodNotFoundError(method, path) return ResolvedRoute( route=routes_tree[method], path_args=path_args, path=path ) def _split_path(path: str) -> Iterable[str]: return path.strip(' /').split('/') PK 2O܋X dataclassesapi/tests/test_app.py# mypy: ignore-errors from dataclasses import dataclass from http import HTTPStatus import pytest from asgi_testclient import TestClient from dataclassesjson import dataclassjson, integer, string from dataclassesapi import MethodType, Route, asgi_app from dataclassesapi.request import Body, Headers, PathArgs, Query, Request from dataclassesapi.response import Body as ResponseBody from dataclassesapi.response import Response @dataclass class FakePathArgs(PathArgs): id: int @dataclass class FakeQuery(Query): query: int @dataclass class FakeHeaders(Headers): x_header: float @dataclass class FakeBody(Body): string: str integer: int @dataclass class FakeRequest(Request): path_args: FakePathArgs query: FakeQuery headers: FakeHeaders body: FakeBody @dataclass class Faked: string: string(max_length=100) integer: integer(minimum=18) @dataclass class FakeResponseBody(ResponseBody): faked: Faked @dataclassjson @dataclass class FakeResponse(Response): body: FakeResponseBody headers: FakeHeaders def fake_controller(req: FakeRequest) -> FakeResponse: return FakeResponse( HTTPStatus.OK, body=FakeResponseBody( faked=Faked(string=req.body.string, integer=req.body.integer) ), headers=FakeHeaders(x_header=req.headers.x_header), ) @pytest.fixture def fake_app(): return asgi_app([Route('/api/{id}', MethodType.GET, fake_controller)]) @pytest.fixture def test_client(fake_app): return TestClient(fake_app) @pytest.mark.asyncio async def test_should_return_not_found(test_client): response = await test_client.get('/not-found') assert response.status_code == HTTPStatus.NOT_FOUND.value assert response.content == b'' assert not response.headers @pytest.mark.asyncio async def test_should_return_method_not_allowed(test_client): response = await test_client.post('/api') assert response.status_code == HTTPStatus.METHOD_NOT_ALLOWED.value assert response.content == b'' assert not response.headers @pytest.mark.asyncio async def test_should_return_bad_request_on_path_arg(test_client): response = await test_client.get('/api/invalid') assert response.status_code == HTTPStatus.BAD_REQUEST.value assert response.json() == { 'type': 'int', 'field': 'id', 'invalid_value': 'invalid', 'error': 'ValueError', } @pytest.mark.asyncio async def test_should_return_bad_request_on_empty_query(test_client): response = await test_client.get('/api/1') assert response.status_code == HTTPStatus.BAD_REQUEST.value assert response.json() == { 'type': 'int', 'field': 'query', 'invalid_value': 'None', } @pytest.mark.asyncio async def test_should_return_bad_request_on_invalid_type_query(test_client): response = await test_client.get('/api/1', params={'query': 'invalid'}) assert response.status_code == HTTPStatus.BAD_REQUEST.value assert response.json() == { 'type': 'int', 'field': 'query', 'invalid_value': 'invalid', 'error': 'ValueError', } @pytest.mark.asyncio async def test_should_return_bad_request_on_empty_header(test_client): response = await test_client.get('/api/1', params={'query': '1'}) assert response.status_code == HTTPStatus.BAD_REQUEST.value assert response.json() == { 'type': 'float', 'field': 'x_header', 'invalid_value': 'None', } @pytest.mark.asyncio async def test_should_return_bad_request_on_invalid_type_header(test_client): response = await test_client.get( '/api/1', params={'query': '1'}, headers={'x-header': 'invalid'} ) assert response.status_code == HTTPStatus.BAD_REQUEST.value assert response.json() == { 'type': 'float', 'field': 'x_header', 'invalid_value': 'invalid', 'error': 'ValueError', } @pytest.mark.asyncio async def test_should_return_ok(test_client): response = await test_client.get( '/api/1', params={'query': '1'}, headers={'x-header': '0.1'}, json={'integer': '1', 'string': 'dataclassesapi'}, ) assert response.status_code == HTTPStatus.OK.value assert response.json() == { 'faked': {'string': 'dataclassesapi', 'integer': 1} } assert dict(response.headers) == { 'x-header': '0.1', 'content-type': 'application/json', 'content-length': '49', } PK2O՝;;#dataclassesapi/tests/test_router.py# mypy: ignore-errors import pytest from dataclassesapi.exceptions import MethodNotFoundError, PathNotFoundError from dataclassesapi.method import MethodType from dataclassesapi.router import Route, route from dataclassesapi.router import router as http_router class TestRouter: @pytest.fixture def method(self): return MethodType.GET @pytest.fixture def controller(self): return lambda: None @pytest.fixture def controller2(self): return lambda: None @pytest.fixture def controller3(self): return lambda: None def test_should_route_path_without_slash(self, method, controller): path = '/dataclasses' router = http_router([Route(path, method, controller)]) resolved = route(router, path, method.value) assert resolved.route.caller is controller def test_should_route_path_with_slash(self, method, controller): path = '/dataclasses/api' router = http_router([Route(path, method, controller)]) resolved = route(router, path, method.value) assert resolved.route.caller is controller def test_should_route_path_with_path_arg_name(self, method, controller): path_pattern = r'/{id}' router = http_router([Route(path_pattern, method, controller)]) path = '/012' resolved = route(router, path, method.value) assert resolved.route.caller is controller assert resolved.path_args == {'id': '012'} def test_should_route_path_with_regex(self, method, controller): path_pattern = r'/{id:\d{3}}' router = http_router([Route(path_pattern, method, controller)]) path = '/012' resolved = route(router, path, method.value) assert resolved.route.caller is controller assert resolved.path_args == {'id': '012'} def test_should_route_path_with_slash_and_regex(self, method, controller): path_pattern = r'/api/{id:\d{3}}' router = http_router([Route(path_pattern, method, controller)]) path = '/api/012' resolved = route(router, path, method.value) assert resolved.route.caller is controller assert resolved.path_args == {'id': '012'} def test_should_route_path_with_regex_and_slash(self, method, controller): path_pattern = r'/{id:\d{3}}/dataclassesapi' router = http_router([Route(path_pattern, method, controller)]) path = '/012/dataclassesapi' resolved = route(router, path, method.value) assert resolved.route.caller is controller def test_should_not_route_path_without_slash(self, method, controller): path_pattern = '/api' router = http_router([Route(path_pattern, method, controller)]) path = '/invalid' with pytest.raises(PathNotFoundError) as exc_info: route(router, path, method) assert exc_info.value.args == (path,) def test_should_not_route_path_with_slash(self, method, controller): path_pattern = '/api/dataclassesapi' router = http_router([Route(path_pattern, method, controller)]) path = '/api/invalid' with pytest.raises(PathNotFoundError) as exc_info: route(router, path, method.value) assert exc_info.value.args == (path,) def test_should_not_route_path_with_regex(self, method, controller): path_pattern = r'/{id:\d{3}}' router = http_router([Route(path_pattern, method, controller)]) path = '/0' with pytest.raises(PathNotFoundError) as exc_info: route(router, path, method.value) assert exc_info.value.args == (path,) def test_should_not_route_path_with_slash_and_regex( self, method, controller ): path_pattern = r'/api/{id:\d{3}}' router = http_router([Route(path_pattern, method, controller)]) path = '/api/0' with pytest.raises(PathNotFoundError) as exc_info: route(router, path, method.value) assert exc_info.value.args == (path,) def test_should_not_route_method(self, method, controller): path = '/api' router = http_router([Route(path, method, controller)]) with pytest.raises(MethodNotFoundError) as exc_info: route(router, path, MethodType.POST.value) assert exc_info.value.args == (MethodType.POST.value, path) def test_should_route_three_paths_with_common_parts( self, method, controller, controller2, controller3 ): path1_pattern = r'/apis/dataclasses/{id:\d{3}}' path2_pattern = r'/apis/dataclasses/{id:\d{3}}/{type}' # path3_pattern = r'/apis/dataclasses/{id:\d{3}}/{type}/{value:.+}' router = http_router( [ Route(path1_pattern, method, controller), Route(path2_pattern, method, controller2), # Route(path3_pattern, method, controller3), ] ) path1 = '/apis/dataclasses/012' path2 = '/apis/dataclasses/013/apis' # path3 = '/apis/dataclasses/014/api/dataclasses' resolved1 = route(router, path1, method.value) resolved2 = route(router, path2, method.value) # resolved3 = route(router, path3, method.value) assert resolved1.route.caller is controller assert resolved2.route.caller is controller2 # assert resolved3.route.caller is controller3 assert resolved1.path_args == {'id': '012'} assert resolved2.path_args == {'id': '013', 'type': 'apis'} # assert resolved3.path_args == { # 'id': '014', # 'type': 'api', # 'value': 'dataclasses', # } PKϺ0O?,,&dataclassesapi-0.2.0.dist-info/LICENSEMIT License Copyright (c) 2019 Diogo Dutra Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HPO$dataclassesapi-0.2.0.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&UrPK!H'dataclassesapi-0.2.0.dist-info/METADATAXmwF_1U A6k6o=HL"i6」 ۳aOrݹ7)QE0svS0 =ifܚqtܞ۱.$r7d?Ht(E1##!d,Wdy&Ҝ1B[G"aNF`pT7jR P$^T(IzKμ!->yé_>9b,!cOJ}4mߏV pg$ ~㾲YJ)9br *v<s$R#q>9ԸHq"|>9T5Cvb%\*!1MS~Dr8KCm+K14Ibt\^Ɖ0ΧJ$"!\|!U->">.%rKsII\,9< "bi>掂 c]-#uX$mY1yc~5gBnvg>7%l t١4 ؖ=۲&b |J~u߂.F▇B ۜ , =kwUsMbyG`5 Zֳme|6(jshv;,!{L$}Xˁ$O$\tsLG7 % Gj;i7=ڷ&4KP SIM3v*JaFqzvEϟl& e5vo-C/.E!ayԉͲ3Mɏ ;+ty3 I= 7TM8K-Aɀ^\09]fǟaHτ : :<ԭ5l]F*S"10h[H"ÌgQ%x(M؁!0{v_t:y.ѱŻA>sU!)Z{7g\xɝ$u8"DJfNA8ZŃE`RIZO4-tC:_'oNVlBb8fXMe=&m.[{Śe7 ~*hm:lW_YڤZgCͿYn L^i5vͣ=uaLY]=DpOXF^â @eH2{?6h̀Y/Q,H5!1 . /v:Z̈́<)c !կ9Ï!V jOݝ1_I=S6'l6+(1n^si12x[kRO">u[iW&ݪi,W ٥ |`1jӞuץo>u',/ nzQ ƽ`p':NWD@`8K춭Wޗ_=SwnoZ VVKiIAlUz(*U,SmePK!H9Qo1%dataclassesapi-0.2.0.dist-info/RECORDI@{pت Qل )v@O?fNlMf.%=O70,Y3Q]9;R +?'p-w(bIߞe wP$$?E4OؘB/pVnҦy'hx][#dIsW|I/@UKoq%'tC;K[]y'Cz'+WIPXM0dBOսtTP۪+SHWCO N#HlX"݅:Ҽ13.7\ b<"OH@2Nv}b_HV+ʓxn XVH2 OB;$y_sCFRJkE WTǚ7D0-jY$bn7m%N^h9Z&٧i\ 2Qh1ZA1(s7cfItYgfyTɲncL8;ʒLd8fI@^d(3y= 7yDB|0HNo4ȺدtDoFȞ/[_ӅTTd;"b)>[(FXpqEİr @yXRPK2O?MY!!dataclassesapi/__init__.pyPK 2O+dͅ Ydataclassesapi/app.pyPK1O*ؿ#dataclassesapi/exceptions.pyPKS 1Oc:#dataclassesapi/method.pyPK1O`dataclassesapi/request.pyPKO1O Tdataclassesapi/response.pyPK2OB= dataclassesapi/router.pyPK 2O܋X )dataclassesapi/tests/test_app.pyPK2O՝;;#;dataclassesapi/tests/test_router.pyPKϺ0O?,,&=Rdataclassesapi-0.2.0.dist-info/LICENSEPK!HPO$Vdataclassesapi-0.2.0.dist-info/WHEELPK!H'?Wdataclassesapi-0.2.0.dist-info/METADATAPK!H9Qo1%J_dataclassesapi-0.2.0.dist-info/RECORDPK a