PK%3OԪapidaora/__init__.py""" ASGI App using dataclasses module for request/response objects """ __version__ = '0.4.1' from apidaora.app import asgi_app from apidaora.method import MethodType from apidaora.request import Request from apidaora.response import Response from apidaora.router import Route __all__ = [ asgi_app.__name__, MethodType.__name__, Route.__name__, Request.__name__, Response.__name__, ] PK3OI apidaora/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 jsondaora.exceptions import DeserializationError from apidaora.exceptions import MethodNotFoundError, PathNotFoundError from apidaora.request import as_request from apidaora.response import AsgiResponse, Response from apidaora.response import as_asgi as as_asgi_response from apidaora.router import Route, route from apidaora.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: logger.error(router, scope['path'], scope['method']) resolved = route(router, scope['path'], scope['method']) except PathNotFoundError: await _send_asgi_response( send, AsgiResponse(HTTPStatus.NOT_FOUND) # type: ignore ) except MethodNotFoundError: await _send_asgi_response( send, AsgiResponse(HTTPStatus.METHOD_NOT_ALLOWED), # type: ignore ) 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( # type: ignore 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*ؿ#apidaora/exceptions.pyclass DataClassesApi(Exception): ... class MethodNotFoundError(DataClassesApi): ... class PathNotFoundError(DataClassesApi): ... PK}2O~[Zapidaora/headers.pyfrom typing import Any, Dict, List, Tuple AsgiPathArgs = Dict[str, Any] AsgiQueryDict = Dict[str, List[str]] AsgiHeaders = List[Tuple[bytes, bytes]] PKS 1Oc:#apidaora/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' PK3O^  apidaora/request.pyfrom collections.abc import Iterable from dataclasses import dataclass from typing import Any, Dict, Optional, Type, TypedDict import orjson from jsondaora import as_typed_dict, asdataclass, jsondaora from .headers import AsgiHeaders, AsgiPathArgs, AsgiQueryDict @jsondaora class PathArgs(TypedDict): ... @jsondaora class Query(TypedDict): ... @jsondaora class Headers(TypedDict): ... @jsondaora class Body(TypedDict): ... @dataclass class Request: path_args: Optional[PathArgs] = None query: Optional[Query] = None headers: Optional[Headers] = None body: Optional[Body] = None 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 asdataclass( # type: ignore dict( path_args=as_typed_dict(path_args, path_args_cls), query=get_query(query_cls, query_dict), headers=get_headers(headers_cls, headers), body=as_typed_dict(orjson.loads(body) if body else {}, body_cls), ), request_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 as_typed_dict(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 as_typed_dict(jsondict, cls) # type: ignore PK3O>z==apidaora/response.pyfrom dataclasses import field from http import HTTPStatus from logging import getLogger from typing import Optional, TypedDict, _TypedDictMeta # type: ignore from jsondaora import dataclass_asjson, jsondaora from apidaora.headers import AsgiHeaders logger = getLogger(__name__) @jsondaora class Headers(TypedDict): ... @jsondaora class Body(TypedDict): ... @jsondaora class Response: status_code: HTTPStatus headers: Headers = field(default_factory=Headers) # type: ignore body: Optional[Body] = None @jsondaora class AsgiResponse: status_code: HTTPStatus headers: AsgiHeaders = field(default_factory=list) body: bytes = b'' def as_asgi(response: Response) -> AsgiResponse: headers_field = type(response).__dataclass_fields__[ # type: ignore 'headers' ] headers: AsgiHeaders = ( as_asgi_headers(response.headers, headers_field.type) if response.headers else [] ) body = dataclass_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( # type: ignore status_code=response.status_code, headers=headers, body=body ) def as_asgi_headers( headers: Optional[Headers], headers_type: _TypedDictMeta ) -> AsgiHeaders: if headers: return [ (field.replace('_', '-').encode(), str(headers[field]).encode()) for field in headers_type.__dataclass_fields__.keys() ] return [] PKĶ2O-L apidaora/router.pyimport re from dataclasses import dataclass from typing import ( Any, Callable, DefaultDict, Dict, Iterable, Optional, Pattern, ) from apidaora.exceptions import MethodNotFoundError, PathNotFoundError from apidaora.method import MethodType from apidaora.request import Request from apidaora.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('/') PK3Oa]]apidaora/tests/test_app.py# mypy: ignore-errors from dataclasses import dataclass from http import HTTPStatus import pytest from asgi_testclient import TestClient from jsondaora import integer, jsondaora, string from apidaora import MethodType, Route, asgi_app from apidaora.request import Body, Headers, PathArgs, Query, Request from apidaora.response import Body as ResponseBody from apidaora.response import Response @jsondaora class FakePathArgs(PathArgs): id: int @jsondaora class FakeQuery(Query): query: int @jsondaora class FakeHeaders(Headers): x_header: float @jsondaora class FakeBody(Body): string: str integer: int @jsondaora class FakeRequest(Request): path_args: FakePathArgs query: FakeQuery headers: FakeHeaders body: FakeBody @jsondaora class Faked: string: string(max_length=100) integer: integer(minimum=18) @jsondaora class FakeResponseBody(ResponseBody): faked: Faked @jsondaora @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': 'apidaora'}, ) assert response.status_code == HTTPStatus.OK.value assert response.json() == {'faked': {'string': 'apidaora', 'integer': 1}} assert dict(response.headers) == { 'x-header': '0.1', 'content-type': 'application/json', 'content-length': '43', } PKŶ2O}a,apidaora/tests/test_router.py# mypy: ignore-errors import pytest from apidaora.exceptions import MethodNotFoundError, PathNotFoundError from apidaora.method import MethodType from apidaora.router import Route, route from apidaora.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}}/apidaora' router = http_router([Route(path_pattern, method, controller)]) path = '/012/apidaora' 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/apidaora' 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?,, apidaora-0.4.1.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!HPOapidaora-0.4.1.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&UrPK!H!apidaora-0.4.1.dist-info/METADATAXmW_1ۉ%ٰP{B6Єi吱4'4hw4,l0ͷ,^q ?̸H|#3ДTHjUw=g=ʹOv4%yƓ)A9ADe$a12H5g$Rdg1sR:U3)W|"\Iԫx>p㑵>b*23ŔG> Dc0Qu fq987l[S:"Q,QhVn ub|7'g:"R\bHcOUy}␓& A~ r'9KȀ4hOJǘ4a$/'sc$Rřk*Y<K*9Ls&"9w.bR@$%ỽʁ0iI![GO)w^A"9.я8*j}=mD/k&¶֒LTgp-]ID.wEV}0K*~Ь=#؜`@18D r,pv@2f 9u _O'Tv4pE#J %)4L|C(fS3%%o"\6w͵{o-  0/f~=h?UWYǪR*3K($ޮ?G^q]ƈ$hvh$StdFp}W7.nhF]OAD(VNSs[*nk Az)!!@CQ,$r"\4뢱tiù.ve)DoPϫ4L R}uXdKZcezW~@q. '֧GQ!Eo?:G2dс4w]-t&!Y[\'+BݷoF%;(\/D6l~0fj@MnkܻBN;d7*9Қ>RoB! CraӂEfȜ!ӯ_O l4i0%]2'Yb3`,wD8p or{ziH4lwt.lUP\^ ʰ;77n<=CT5e8D"L +gQ(\^.`$`; pe4˻4ᅕGz=_'oGVa| ZySo!%S5w嗖uk}nmO7F74ãbK/?-9>e[tˉaA?Chf%\Sx! _ʪGfkw n~fX a!@|aXCDCtZv_"{H S;7E ^5v9Xycj c9=z4%o5Q1.UE÷8˔:걊u]Uf&-^oƇbjt@˶/\0uu[jN#ev~6AW|j@#` 0%7lFo;'ui<;dV'=ίOZ7sO7ƛ/tk*BҺ^pNX4!67~fsiO]z4z}Y*cv9ա(# ZBtʵOPK!H.apidaora-0.4.1.dist-info/RECORD}ɮP}E/2 *2 APM'Fn:7Rk5]+ u5|"}{\*E<L:zfpoüGHVa&!Dώ6A#U^ g!Up=ub<. M/̰n3Eid9 ';-b.Τ:-`̽vWBnîI,S>qA%bWábQV+N9ΰ^3ތ*HUXúO(U;b vkfI. Y7$ڡز\ +U+bcj'Uf Bo+tMu|յN_"a*cGMרAlFl8&iU{(_lFPd7%@'18ٔ>::28(s;G 5͋Ϳh[+.VRW}OۙPAf Y><PK%3OԪapidaora/__init__.pyPK3OI apidaora/app.pyPK1O*ؿ#apidaora/exceptions.pyPK}2O~[Zapidaora/headers.pyPKS 1Oc:#~apidaora/method.pyPK3O^  apidaora/request.pyPK3O>z==apidaora/response.pyPKĶ2O-L F"apidaora/router.pyPK3Oa]]V-apidaora/tests/test_app.pyPKŶ2O}a,>apidaora/tests/test_router.pyPKϺ0O?,, 7Uapidaora-0.4.1.dist-info/LICENSEPK!HPOYapidaora-0.4.1.dist-info/WHEELPK!H!-Zapidaora-0.4.1.dist-info/METADATAPK!H.bbapidaora-0.4.1.dist-info/RECORDPK2e