PKtYDO)%]]apidaora/__init__.py""" ASGI App using dataclasses module for request/response objects """ __version__ = '0.6.0' from apidaora.content import ContentType from apidaora.core.response import ( HTMLResponse, JSONResponse, PlainResponse, Response, ) from apidaora.method import MethodType from apidaora.openapi.app import operations_app as appdaora from apidaora.openapi.app import spec_app as appdaora_spec from apidaora.openapi.parameters import header_param from apidaora.openapi.path_decorator.base import path from apidaora.openapi.request import JSONRequestBody __all__ = [ ContentType.__name__, MethodType.__name__, path.__name__, appdaora.__name__, appdaora_spec.__name__, HTMLResponse.__name__, JSONResponse.__name__, PlainResponse.__name__, Response.__name__, header_param.__name__, JSONRequestBody.__name__, ] PK5O^0Oapidaora/content.pyfrom enum import Enum class ContentType(Enum): APPLICATION_JSON = 'application/json' TEXT_HTML = 'text/html; charset=utf-8' TEXT_PLAIN = 'text/plain' PK66OVh|apidaora/exceptions.pyclass APIDaoraError(Exception): ... class MethodNotFoundError(APIDaoraError): ... class PathNotFoundError(APIDaoraError): ... class InvalidPathParams(APIDaoraError): ... 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' PK5Oapidaora/core/__init__.pyPKkBO+( apidaora/core/app.pyimport asyncio from http import HTTPStatus from logging import getLogger from typing import Any, Awaitable, Callable, Coroutine, Dict, Iterable from urllib import parse from jsondaora.exceptions import DeserializationError from ..exceptions import MethodNotFoundError, PathNotFoundError from .request import as_request from .response import AsgiResponse, JSONResponse, Response from .response import as_asgi as as_asgi_response from .router import Route, route from .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.exception(f"DeserializationError {error.message}") response = JSONResponse(body={'error': error.dict}) response.__status_code__ = ( # type: ignore HTTPStatus.BAD_REQUEST ) asgi_response = as_asgi_response(response) 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: if asyncio.iscoroutine(response): response = await response # type: ignore 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}) PK*76OTH??apidaora/core/headers.pyfrom typing import Any, Dict, List, Optional, Tuple, TypedDict from jsondaora import jsondaora from ..content import ContentType AsgiPathArgs = Dict[str, Any] AsgiQueryDict = Dict[str, List[str]] AsgiHeaders = List[Tuple[bytes, bytes]] @jsondaora class Headers(TypedDict): content_type: Optional[ContentType] PK86OHlj  apidaora/core/request.pyfrom collections.abc import Iterable from dataclasses import dataclass from logging import getLogger from typing import Any, Dict, Optional, Type, TypedDict import orjson from jsondaora import ( as_typed_dict, as_typed_dict_field, asdataclass, jsondaora, ) from ..content import ContentType from .headers import AsgiHeaders, AsgiPathArgs, AsgiQueryDict, Headers logger = getLogger(__name__) @jsondaora class PathArgs(TypedDict): ... @jsondaora class Query(TypedDict): ... @jsondaora class Body(TypedDict): ... @dataclass class Request: path_args: PathArgs query: Query headers: Headers body: Optional[Body] def as_request( request_cls: Type[Request], body: bytes, path_args: AsgiPathArgs = {}, query_dict: AsgiQueryDict = {}, headers: AsgiHeaders = [], ) -> 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) request_path_args = as_typed_dict(path_args, path_args_cls) request_query = get_query(query_cls, query_dict) request_headers = get_headers(headers_cls, headers) content_type = request_headers.get('content_type') parsed_body: Any = None if content_type is None or content_type is ContentType.APPLICATION_JSON: parsed_body = orjson.loads(body) if body else {} parsed_body = as_typed_dict_field(parsed_body, 'body', body_cls) else: parsed_body = body.decode() return asdataclass( # type: ignore dict( path_args=request_path_args, query=request_query, headers=request_headers, body=parsed_body if parsed_body else None, ), request_cls, skip_fields=('body',), ) 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 PK-96O7( apidaora/core/response.pyfrom dataclasses import dataclass, field from http import HTTPStatus from logging import getLogger from typing import ( # type: ignore Any, Dict, Optional, TypedDict, _TypedDictMeta, ) from jsondaora import dataclass_asjson, jsondaora from ..content import ContentType from .headers import AsgiHeaders, Headers logger = getLogger(__name__) @jsondaora class Body(TypedDict): error: Dict[str, Any] @dataclass class Response: body: Body headers: Headers = field(default_factory=Headers) # type: ignore Response.__content_type__: Optional[ContentType] = None # type: ignore Response.__status_code__: HTTPStatus = HTTPStatus.OK # type: ignore @dataclass class JSONResponse(Response): __content_type__ = ContentType.APPLICATION_JSON @dataclass class HTMLResponse(Response): __content_type__ = ContentType.TEXT_HTML @dataclass class PlainResponse(Response): __content_type__ = ContentType.TEXT_PLAIN @dataclass 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: bytes = b'' if response.body: if ( type(response).__content_type__ # type: ignore == JSONResponse.__content_type__ ): body = dataclass_asjson(response.body) elif not isinstance(response.body, bytes): body = str(response.body).encode() if body: if type(response).__content_type__: # type: ignore headers.append( ( b'Content-Type', type( response ).__content_type__.value.encode(), # type: ignore ) ) headers.append((b'Content-Length', str(len(body)).encode())) return AsgiResponse( status_code=response.__status_code__, # type: ignore headers=headers, body=body, ) def as_asgi_headers( headers: Optional[Headers], headers_type: _TypedDictMeta ) -> AsgiHeaders: if headers: return [ (field.replace('_', '-').encode(), str(value).encode()) for field, value in headers.items() ] return [] PKkBOk: apidaora/core/router.pyimport re from dataclasses import dataclass from typing import ( Any, Callable, DefaultDict, Dict, Iterable, Optional, Pattern, ) from ..exceptions import MethodNotFoundError, PathNotFoundError from ..method import MethodType from .request import Request from .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 PATH_RE = re.compile(r'\{(?P[^/:]+)(:(?P[^/:]+))?\}') def router(routes: Iterable[Route]) -> RoutesTree: routes_tree = RoutesTree() for route in routes: path_pattern_parts = split_path(route.path_pattern.rstrip('/ ')) routes_tree_tmp = routes_tree for path_pattern_part in path_pattern_parts: match = PATH_RE.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('/') PK56Oapidaora/core/tests/test_app.py# mypy: ignore-errors from http import HTTPStatus import pytest from asgi_testclient import TestClient from jsondaora import integer, jsondaora, string from apidaora import MethodType from apidaora.core.app import asgi_app from apidaora.core.headers import Headers from apidaora.core.request import Body as RequestBody from apidaora.core.request import PathArgs, Query, Request from apidaora.core.response import Body as ResponseBody from apidaora.core.response import JSONResponse from apidaora.core.router import Route @jsondaora class FakePathArgs(PathArgs): id: int @jsondaora class FakeQuery(Query): query: int @jsondaora class FakeHeaders(Headers): x_header: float @jsondaora class FakeBody(RequestBody): 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 class FakeResponse(JSONResponse): body: FakeResponseBody headers: FakeHeaders def fake_controller(req: FakeRequest) -> FakeResponse: return FakeResponse( 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() == { 'error': { 'type': 'int', 'field': 'id', 'invalid_value': 'invalid', 'name': 'ValueError', 'cls': 'FakePathArgs', } } @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() == { 'error': { 'type': 'int', 'field': 'query', 'invalid_value': None, 'cls': 'FakeQuery', 'name': 'ParameterNotFoundError', } } @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() == { 'error': { 'type': 'int', 'field': 'query', 'invalid_value': 'invalid', 'name': 'ValueError', 'cls': 'FakeQuery', } } @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() == { 'error': { 'type': 'float', 'field': 'x_header', 'invalid_value': None, 'cls': 'FakeHeaders', 'name': 'ParameterNotFoundError', } } @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() == { 'error': { 'type': 'float', 'field': 'x_header', 'invalid_value': 'invalid', 'name': 'ValueError', 'cls': 'FakeHeaders', } } @pytest.mark.asyncio async def test_should_return_bad_request_on_empty_body(test_client): response = await test_client.get( '/api/1', params={'query': '1'}, headers={'x-header': '0.1'} ) assert response.status_code == HTTPStatus.BAD_REQUEST.value assert response.json() == { 'error': { 'type': 'str', 'field': 'string', 'invalid_value': None, 'name': 'ParameterNotFoundError', 'cls': 'FakeBody', } } @pytest.mark.asyncio async def test_should_return_bad_request_on_invalid_type_body(test_client): response = await test_client.get( '/api/1', params={'query': '1'}, headers={'x-header': '0.1'}, json={'string': 'str', 'integer': 'str'}, ) assert response.status_code == HTTPStatus.BAD_REQUEST.value assert response.json() == { 'error': { 'type': 'int', 'field': 'integer', 'invalid_value': 'str', 'name': 'ValueError', 'cls': 'FakeBody', } } @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', } PKR{5O4~"apidaora/core/tests/test_router.py# mypy: ignore-errors import pytest from apidaora.core.router import Route, route from apidaora.core.router import router as http_router from apidaora.exceptions import MethodNotFoundError, PathNotFoundError from apidaora.method import MethodType 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', # } PKx5Oapidaora/openapi/__init__.pyPKJ8Okg__apidaora/openapi/app.pyfrom typing import Any, Callable, Sequence from ..core.app import AsgiCallable, asgi_app def operations_app(operations: Sequence[Callable[..., Any]]) -> AsgiCallable: routes = [] for operation in operations: routes.append(operation.partial_path.route) # type: ignore return asgi_app(routes) def spec_app() -> None: ... PK8Ov1;ssapidaora/openapi/parameters.pyfrom dataclasses import dataclass from enum import Enum from typing import Any, Optional, Type class ParameterType(Enum): HEADER = 'header' @dataclass class Parameter: in_: ParameterType name: Optional[str] schema: Type[Any] required: bool = True description: str = '' def header_param(schema: Type[Any], name: Optional[str] = None) -> Type[Any]: schema_ = schema name_ = name @dataclass class HeaderParameter(Parameter): ... HeaderParameter.name = name_ HeaderParameter.schema = schema_ HeaderParameter.in_ = ParameterType.HEADER return HeaderParameter PK76Oupmapidaora/openapi/request.pyfrom typing import Any, Dict from jsondaora import jsondaora from ..content import ContentType @jsondaora class JSONRequestBody: content: Dict[str, Any] JSONRequestBody.__content_type__: ContentType = ContentType.APPLICATION_JSON # type: ignore PK5Oapidaora/openapi/response.pyPKǥ8OWapidaora/openapi/security.pyfrom enum import Enum from typing import TypedDict from jsondaora import jsondaora @jsondaora class Security(TypedDict): ... class SecurityType(Enum): OAUTH2 = 'OAuth2' PKͦ8Ofapidaora/openapi/spec.py# type: ignore from http import HTTPStatus from typing import ( Any, Dict, Iterable, Optional, Type, TypedDict, Union, _GenericAlias, ) from jsondaora import jsondaora from ..content import ContentType from ..method import MethodType from .parameters import Parameter from .security import Security, SecurityType @jsondaora class OpenAPIMethod(TypedDict): parameters: Iterable[Parameter] class Response(TypedDict): class Content(TypedDict): schema: Type[Any] content: Dict[ContentType, Content] responses: Dict[HTTPStatus, Response] security: Dict[SecurityType, Security] request_body: Optional[Type[Any]] = None operation_id: Optional[str] = None ParameterName = str SchemaName = str PathPattern = str @jsondaora class OpenAPISpec(TypedDict): class Server(TypedDict): url: str description: str class Info(TypedDict): version: str title: str class Components(TypedDict): parameters: Dict[ParameterName, Parameter] security_schemes: Dict[SecurityType, Security] schemas: Dict[SchemaName, Type[Any]] openapi: str info: Info server: Server paths: Dict[PathPattern, Dict[MethodType, OpenAPIMethod]] def get_operation_request_bodies(body_type): operation_bodies = {} if isinstance(body_type, _GenericAlias) and body_type.__origin__ is Union: for union_type in body_type.__args__: if not (union_type is type(None) or union_type is None): # noqa content_type = union_type.__content_type__ if isinstance(content_type, ContentType): operation_bodies[content_type] = union_type else: content_type = body_type.__content_type__ if isinstance(content_type, ContentType): operation_bodies[content_type] = body_type return operation_bodies PKg8Oz44'apidaora/openapi/path_decorator/base.pyfrom logging import getLogger from typing import Any, Callable, Dict, Optional, Tuple, Type from ...core.request import Body as RequestBody from ...core.response import Response from ...core.router import Route from ...method import MethodType from ..spec import OpenAPIMethod from .factories.base import ( make_core_operation, make_json_fields, query, request, ) from .factories.headers import request_headers from .factories.path_args import path_args from .partial_path import PartialPath logger = getLogger(__name__) RequestParams = Dict[Tuple[str, Type[Any]], Type[Any]] def path(pattern: str, method: MethodType) -> Any: def operation_wrapper( operation: Callable[..., Response] ) -> Callable[..., Response]: operation_annotations = { k: v for k, v in operation.__annotations__.items() if k not in ('return', 'body') } json_fields = make_json_fields(operation_annotations) OperationPathArgs = path_args(pattern, operation_annotations) headers_names: Dict[str, str] = {} OperationRequestHeaders = request_headers( operation_annotations, headers_names ) OperationQuery = query( OperationPathArgs, headers_names, operation_annotations ) OperationRequestBody = ( operation.__annotations__.get('body') or Optional[RequestBody] ) OperationRequest = request( OperationPathArgs, OperationQuery, OperationRequestHeaders, OperationRequestBody, ) OperationResponse = operation.__annotations__['return'] core_operation = make_core_operation( OperationRequest, OperationResponse, operation, json_fields, headers_names, ) route = Route(pattern, method, core_operation) partial_path = PartialPath(method=OpenAPIMethod(), route=route) core_operation.partial_path = partial_path return core_operation return operation_wrapper PKǥ8Om{/apidaora/openapi/path_decorator/partial_path.pyfrom dataclasses import dataclass from ...core.router import Route from ..spec import OpenAPIMethod @dataclass class PartialPath: method: OpenAPIMethod route: Route PK8O'a~v v 1apidaora/openapi/path_decorator/factories/base.pyfrom dataclasses import dataclass from logging import getLogger from typing import ( # type: ignore Any, Callable, Dict, Iterable, Optional, Set, Type, _TypedDictMeta, ) from jsondaora import jsondaora from jsondaora.schema import JsonField from ....core.headers import Headers from ....core.request import Body as RequestBody from ....core.request import PathArgs, Query, Request from ....core.response import Response from ..partial_path import PartialPath logger = getLogger(__name__) def make_json_fields(operation_annotations: Dict[str, Any]) -> Set[str]: json_fields = set() for param_name, param_type in operation_annotations.items(): if isinstance(param_type, type) and issubclass(param_type, JsonField): json_fields.add(param_name) return json_fields def query( path_args_type: Type[PathArgs], headers_names: Dict[str, str], operation_annotations: Dict[str, Any], ) -> Type[Query]: query_annotations = { k: v for k, v in operation_annotations.items() if k not in path_args_type.__annotations__ and k not in headers_names.keys() and k not in headers_names.values() } query_type = _TypedDictMeta( 'OperationQuery', (Query,), {'__annotations__': query_annotations} ) query_type: Type[Query] = jsondaora(query_type) return query_type def request( path_args_type: PathArgs, query_type: Query, headers_type: Headers, body_type: RequestBody, ) -> Type[Request]: @dataclass class OperationRequest(Request): path_args: path_args_type # type: ignore query: query_type # type: ignore headers: headers_type # type: ignore body: body_type return OperationRequest def make_core_operation( request_type: Request, request_response: Response, operation: Callable[..., Response], json_fields: Iterable[JsonField], headers_names: Dict[str, str], partial_path: Optional[PartialPath] = None, ) -> Type[Any]: def core_operation(req: request_type) -> request_response: operation_kwargs: Dict[str, Any] = {} operation_kwargs.update(req.path_args) operation_kwargs.update(req.query) operation_kwargs.update( (operation_name, req.headers[name]) # type: ignore for name, operation_name in headers_names.items() ) operation_kwargs.update( (field, operation_kwargs[field].value) for field in json_fields ) if req.body: operation_kwargs['body'] = req.body return operation(**operation_kwargs) return core_operation PK8Oa;apidaora/openapi/path_decorator/factories/core_operation.pyfrom logging import getLogger from typing import Any, Callable, Dict, Iterable, Optional, Type from jsondaora.schema import JsonField from ....core.request import Request from ....core.response import Response from ..partial_path import PartialPath logger = getLogger(__name__) def make_core_operation( request_type: Request, request_response: Response, operation: Callable[..., Response], json_fields: Iterable[JsonField], headers_names: Dict[str, str], partial_path: Optional[PartialPath] = None, ) -> Type[Any]: def core_operation(req: request_type) -> request_response: operation_kwargs: Dict[str, Any] = {} operation_kwargs.update(req.path_args) operation_kwargs.update(req.query) operation_kwargs.update( (operation_name, req.headers[name]) # type: ignore for name, operation_name in headers_names.items() ) operation_kwargs.update( (field, operation_kwargs[field].value) for field in json_fields ) if req.body: operation_kwargs['body'] = req.body return operation(**operation_kwargs) return core_operation PK8O 4apidaora/openapi/path_decorator/factories/headers.pyfrom logging import getLogger from typing import ( # type: ignore Any, Dict, Optional, Set, Tuple, Type, Union, _GenericAlias, _TypedDictMeta, ) from jsondaora import jsondaora from ....core.headers import Headers from ...parameters import Parameter, ParameterType logger = getLogger(__name__) def request_headers( operation_annotations: Dict[str, Any], headers_names: Dict[str, str] ) -> Type[Headers]: headers_fields = set() optional_fields: Set[Tuple[str, Type[Any]]] = set() annotations: Dict[str, Any] = {} attrs: Dict[str, Any] = {} for param_name, param_type in operation_annotations.items(): if ( isinstance(param_type, type) and issubclass(param_type, Parameter) and param_type.in_ is ParameterType.HEADER ): parse_header_param( param_type, param_name, headers_names, optional_fields, headers_fields, ) for k, v in headers_fields: if (k, v) in optional_fields: v = Optional[v] attrs[k] = None annotations[k] = v attrs['__annotations__'] = annotations request_headers_type = _TypedDictMeta('OperationHeaders', (), attrs) request_headers_type: Type[Headers] = jsondaora(request_headers_type) return request_headers_type def parse_header_param( param_type, operation_param_name, headers_names, optional_fields, headers_fields, ) -> None: param_type_name = ( param_type.name.replace('-', '_') if param_type.name else None ) field_name = param_type_name or operation_param_name if param_type_name: headers_names[param_type_name] = operation_param_name field_type = get_field_type(param_type) field = (field_name, field_type) if not param_type.required: optional_fields.add(field) headers_fields.add(field) def get_field_type(param_type): field_type = param_type.schema if ( isinstance(field_type, _GenericAlias) and field_type.__origin__ is Union ): field_types = set() for union_type in field_type.__args__: if union_type is type(None) or union_type is None: # noqa param_type.required = False elif ( issubclass(param_type, Parameter) and param_type.in_ is ParameterType.HEADER ): field_types.add(union_type) union_types = [f_type.__name__ for f_type in field_types] if not param_type.required: union_types.append('None') field_type = Union[eval(', '.join(union_types))] return field_type PKkBO'7A6apidaora/openapi/path_decorator/factories/path_args.pyfrom logging import getLogger from typing import Any, Dict, Type, _TypedDictMeta # type: ignore from jsondaora import jsondaora from ....core.request import PathArgs from ....core.router import PATH_RE, split_path from ....exceptions import InvalidPathParams logger = getLogger(__name__) def path_args( pattern: str, operation_annotations: Dict[str, Any] ) -> Type[PathArgs]: path_args_matchs = map(lambda p: PATH_RE.match(p), split_path(pattern)) path_args_names = set( match.groupdict()['name'] for match in path_args_matchs if match ) path_args_annotations = { k: v for k, v in operation_annotations.items() if k in path_args_names } if len(path_args_annotations) != len(path_args_names): missing_args = ', '.join( set(path_args_annotations.keys()).difference(path_args_names) ) invalid_args = ', '.join( set(path_args_names).difference(path_args_annotations.keys()) ) raise InvalidPathParams( f'Missing args: {missing_args}. Invalid args: {invalid_args}' ) if path_args_annotations: OperationPathArgs = _TypedDictMeta( 'OperationPathArgs', (PathArgs,), {'__annotations__': path_args_annotations}, ) OperationPathArgs: Type[PathArgs] = jsondaora(OperationPathArgs) else: OperationPathArgs = PathArgs return OperationPathArgs PKS8O.Pvk//<apidaora/openapi/tests/test_openapi_app_scalar_parameters.pyfrom http import HTTPStatus import pytest from asgi_testclient import TestClient from jsondaora import jsondaora from apidaora import MethodType, Response, appdaora, header_param, path @pytest.fixture def hello_fake(): return 'Hello Fake!' @jsondaora class StrResponse(Response): body: str @pytest.fixture def simple_client(hello_fake): @path('/test/', MethodType.GET) def operation() -> StrResponse: return StrResponse(body=hello_fake) app = appdaora([operation]) return TestClient(app) @pytest.mark.asyncio async def test_should_get_response_without_arguments( simple_client, hello_fake ): response = await simple_client.get('/test/') assert response.status_code == HTTPStatus.OK.value assert response.content == hello_fake.encode() @pytest.mark.asyncio async def test_should_get_response_without_arguments_and_slash( simple_client, hello_fake ): response = await simple_client.get('/test') assert response.status_code == HTTPStatus.OK.value assert response.content == hello_fake.encode() @pytest.mark.asyncio async def test_should_get_response_with_one_scalar_path_parameter(): @path('/test/{param}', MethodType.GET) def operation(param: int) -> StrResponse: return StrResponse(body=f'{param}: {type(param).__name__}') client = TestClient(appdaora([operation])) response = await client.get('/test/01') assert response.status_code == HTTPStatus.OK.value assert response.content == b'1: int' @pytest.mark.asyncio async def test_should_get_response_with_two_scalar_path_parameters(): @path('/test/{param}/{param2}', MethodType.GET) def operation(param: int, param2: float) -> StrResponse: return StrResponse( body=f'{param}: {type(param).__name__}, {param2}: {type(param2).__name__}' ) client = TestClient(appdaora([operation])) response = await client.get('/test/01/00000.1') assert response.status_code == HTTPStatus.OK.value assert response.content == b'1: int, 0.1: float' @pytest.mark.asyncio async def test_should_get_response_with_one_scalar_path_parameter_and_one_query_parameter(): @path('/test/{param}', MethodType.GET) def operation(param: int, param2: float) -> StrResponse: return StrResponse( body=f'{param}: {type(param).__name__}, {param2}: {type(param2).__name__}' ) client = TestClient(appdaora([operation])) response = await client.get('/test/01/?param2=00000.1') assert response.status_code == HTTPStatus.OK.value assert response.content == b'1: int, 0.1: float' @pytest.mark.asyncio async def test_should_get_response_with_one_scalar_path_parameter_and_two_query_parameter(): @path('/test/{param}', MethodType.GET) def operation(param: int, param2: float, param3: str) -> StrResponse: return StrResponse( body=( f'{param}: {type(param).__name__}, ' f'{param2}: {type(param2).__name__}, ' f'{param3}: {type(param3).__name__}' ) ) client = TestClient(appdaora([operation])) response = await client.get('/test/01/?param2=00000.1¶m3=test') assert response.status_code == HTTPStatus.OK.value assert response.content == b'1: int, 0.1: float, test: str' @pytest.mark.asyncio async def test_should_get_response_with_scalar_parameters_with_one_path_and_one_query_and_one_header(): @path('/test/{param}', MethodType.GET) def operation( param: int, param2: float, param3: header_param(schema=float, name='x-param3'), ) -> StrResponse: return StrResponse( body=( f'{param}: {type(param).__name__}, ' f'{param2}: {type(param2).__name__}, ' f'{param3}: {type(param3).__name__}' ) ) client = TestClient(appdaora([operation])) response = await client.get( '/test/01/?param2=00000.1', headers={'x-param3': '-000.1'} ) assert response.status_code == HTTPStatus.OK.value assert response.content == b'1: int, 0.1: float, -0.1: float' PKϺ0O?,, apidaora-0.6.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!HPOapidaora-0.6.0.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&UrPK!H!apidaora-0.6.0.dist-info/METADATAXS:Buw7hbn)< Ul%Q-_K&L=G_M 3߶8GLѐ*e#uLc򐊌Zp5Zyl鑽wd/MI.y2#('LXyTd$cL*7c2dDL@I뽈+Jug\ *aHʌ<`OO>[{#\9@氘#!h 潚:ʷ,69Kv }((gLAbǯCH}tO90y\G$7,qb9`7,i ɹ*H="9h@8[hH"$к HN41#Mf9D ?q.jA32C'8uT*ER$燍ȿYFTdeF;< K|J&R-#P3Pb{"dyƦ+93"fσ@lJv%,x澽9ڮ6ǥ֎q(gIʒC)<ޜ 03m ;%4 qq0<2(a(NqAqzO(;ׯk{t,@3}zOUh֎[lI2@xzXhfEttubC4Qu%$X h/_X!)O ׫Q)_S g-q1 ׋XZLĭ8S^ Rd3S<}r@XTSIJԿeORuBzaē S t;E4 ޽MIR&eD#n3)q ɴő;$3ԆOZz6g?( CM nx @х Y,?٬Aq*V/~&8<~{4)Ԍ>0\663 ò!JȜCNħtPJ9X,,y> 3;HE`?#p8$ }m(A7Uk#Y6l1ہd.u!Ue8D"\$;YdQ؈-U.d,?`A%Cy$4Mr!=;[^p+033}#T%35eEjVo.Y+`Gr > Iuɧ! pprƲ~~C[]]Ek_Cc@JD{$mhȲ+Y+}K"LipJ3A6yπL U{I]8ы::MGVh2Ґ<ZCZp&eqpuK.Y/%5zMk٣7t`T5\t0ZS_ï[cp):k2&Yg#C<+1{u?TIPWv"+w:Ɓ8T|"LzX{WM0ڄ++vtEKdoO/֭xn8CQY+ݘ^H叆^$Z}ڕ/[Ai:F رH "WLtk>-8l<켊! _٪GS;2\elc !,_$*0>XTivg~Y~=0%- /xIrqI;$=c:I՚0l<|SsgWl{ђhjmȼmn|/FuFjmYwR8X2#Swg2ظ}6]-1./B1T,n=G|B!cHz[ގe͆CLs!Be:Op1gk_HkS% t= T3dn>vhk<`n 7$ _>{9n=/t|ƿ~cs&PK!HYD8" apidaora-0.6.0.dist-info/RECORDGF-hؠBDBdĆ"'DЯ73a^X  =o]m|)F-А(lȷ-;KꔏQtXpw'(fܠjլmyJh>pyUxYdJw*?@gtc6ÆEZ]‹;hS`x l=*ŸB/ZyoH>N{\axQI/J 8r9^&w̎Sm}Ypav)Ύ8IK&Jwr<ˍ . )̤͢BE5`*k#d$=sX =,O8]}I>a!:nb/5i|yƕwaOS2l=kʦ`G&]AB  ԇ쓡[mM6D v"JGK,%.{:&4,NcoxJݞڋb*3:Gڥ{3gBב߮BI⽽ZnMX?}I<(cFKr3[|e(0JHq2 -"y߈T(fKm47?˨v;TFsh3yǘy.̽r]s|uAp63'c̓˅zS-E\۫uK[=@%?R?s;HH'39:g/0^}9}&⸬ܚ !CM}1.UNryѡ{vyTnDscB0$f~}>#K o1f]-4!ϙf]u6b`8Ym= ~1r!M c4]}KsG&qX0{HFÅKL˪.Vlh0  Ľw2i"vϾHaIvWVy4\ O2$b싅P@RBnׯۚ$(/I۲s9 b)jx4;z"$/4?B1hwdnd &<ё;Ï)R狠϶rשgL2 .K*t *@WWĞ /qO6Q,,%xQu4"NZ;?IG`"p=N8#8bwEW س1 VmL;ábޮq2Fog B`@} v߰'"Yַ+y+Kb(ʐX UX'g' .a:\6z]o9eGUg 菿PKtYDO)%]]apidaora/__init__.pyPK5O^0Oapidaora/content.pyPK66OVh|dapidaora/exceptions.pyPKS 1Oc:#Xapidaora/method.pyPK5Ojapidaora/core/__init__.pyPKkBO+( apidaora/core/app.pyPK*76OTH??apidaora/core/headers.pyPK86OHlj  apidaora/core/request.pyPK-96O7( B!apidaora/core/response.pyPKkBOk: t+apidaora/core/router.pyPK56Om6apidaora/core/tests/test_app.pyPKR{5O4~"xNapidaora/core/tests/test_router.pyPKx5Odapidaora/openapi/__init__.pyPKJ8Okg__ eapidaora/openapi/app.pyPK8Ov1;ssfapidaora/openapi/parameters.pyPK76OupmPiapidaora/openapi/request.pyPK5Ojapidaora/openapi/response.pyPKǥ8OWjapidaora/openapi/security.pyPKͦ8Ofkapidaora/openapi/spec.pyPKg8Oz44'usapidaora/openapi/path_decorator/base.pyPKǥ8Om{/{apidaora/openapi/path_decorator/partial_path.pyPK8O'a~v v 1|apidaora/openapi/path_decorator/factories/base.pyPK8Oa;apidaora/openapi/path_decorator/factories/core_operation.pyPK8O 4apidaora/openapi/path_decorator/factories/headers.pyPKkBO'7A6apidaora/openapi/path_decorator/factories/path_args.pyPKS8O.Pvk//<apidaora/openapi/tests/test_openapi_app_scalar_parameters.pyPKϺ0O?,, ;apidaora-0.6.0.dist-info/LICENSEPK!HPOapidaora-0.6.0.dist-info/WHEELPK!H!1apidaora-0.6.0.dist-info/METADATAPK!HYD8" Lapidaora-0.6.0.dist-info/RECORDPK#