diff --git a/Makefile b/Makefile index d6d9a17f..9670ec61 100644 --- a/Makefile +++ b/Makefile @@ -3,18 +3,18 @@ check: lint test SOURCE_FILES=spectree tests examples setup.py install: - pip install -e .[email,flask,falcon,starlette,dev] + pip install -e .[email,quart,flask,falcon,starlette,dev] import_test: pip install -e .[email] - for module in flask falcon starlette; do \ + for module in flask quart falcon starlette; do \ pip install -U $$module; \ bash -c "python tests/import_module/test_$${module}_plugin.py" || exit 1; \ pip uninstall $$module -y; \ done test: import_test - pip install -U -e .[email,flask,falcon,starlette] + pip install -U -e .[email,flask,quart,falcon,starlette] pytest tests -vv -rs doc: diff --git a/examples/quart_demo.py b/examples/quart_demo.py new file mode 100644 index 00000000..341f35a4 --- /dev/null +++ b/examples/quart_demo.py @@ -0,0 +1,108 @@ +from enum import Enum +from random import random + +from pydantic import BaseModel, Field +from quart import Quart, abort, jsonify, request +from quart.views import MethodView + +from spectree import Response, SpecTree + +app = Quart(__name__) +api = SpecTree("quart") + + +class Query(BaseModel): + text: str = "default query strings" + + +class Resp(BaseModel): + label: int + score: float = Field( + ..., + gt=0, + lt=1, + ) + + +class Data(BaseModel): + uid: str + limit: int = 5 + vip: bool + + class Config: + schema_extra = { + "example": { + "uid": "very_important_user", + "limit": 10, + "vip": True, + } + } + + +class Language(str, Enum): + en = "en-US" + zh = "zh-CN" + + +class Header(BaseModel): + Lang: Language + + +class Cookie(BaseModel): + key: str + + +@app.route( + "/api/predict//", methods=["POST"] +) +@api.validate( + query=Query, json=Data, resp=Response("HTTP_403", HTTP_200=Resp), tags=["model"] +) +def predict(source, target): + """ + predict demo + + demo for `query`, `data`, `resp`, `x` + + query with + ``http POST ':8000/api/predict/zh/en?text=hello' uid=xxx limit=5 vip=false `` + """ + print(f"=> from {source} to {target}") # path + print(f"JSON: {request.json}") # Data + print(f"Query: {request.args}") # Query + if random() < 0.5: + abort(403) + + return jsonify(label=int(10 * random()), score=random()) + + +@app.route("/api/header", methods=["POST"]) +@api.validate( + headers=Header, cookies=Cookie, resp=Response("HTTP_203"), tags=["test", "demo"] +) +async def with_code_header(): + """ + demo for JSON with status code and header + + query with ``http POST :8000/api/header Lang:zh-CN Cookie:key=hello`` + """ + return jsonify(language=request.headers.get("Lang")), 203, {"X": 233} + + +class UserAPI(MethodView): + @api.validate(json=Data, resp=Response(HTTP_200=Resp), tags=["test"]) + async def post(self): + return jsonify(label=int(10 * random()), score=random()) + # return Resp(label=int(10 * random()), score=random()) + + +if __name__ == "__main__": + """ + cmd: + http :8000/api/user uid=12 limit=1 vip=false + http ':8000/api/predict/zh/en?text=hello' vip=true uid=aa limit=1 + http POST :8000/api/header Lang:zh-CN Cookie:key=hello + """ + app.add_url_rule("/api/user", view_func=UserAPI.as_view("user_id")) + api.register(app) + app.run(port=8000) diff --git a/setup.py b/setup.py index 7de48cff..160239bd 100644 --- a/setup.py +++ b/setup.py @@ -44,6 +44,7 @@ extras_require={ "email": ["pydantic[email]>=1.2"], "flask": ["flask"], + "quart": ["quart"], "falcon": ["falcon>=3.0.0"], "starlette": ["starlette[full]"], "dev": [ diff --git a/spectree/plugins/__init__.py b/spectree/plugins/__init__.py index 6796218d..c5222258 100644 --- a/spectree/plugins/__init__.py +++ b/spectree/plugins/__init__.py @@ -9,6 +9,7 @@ PLUGINS = { "base": Plugin(".base", __name__, "BasePlugin"), "flask": Plugin(".flask_plugin", __name__, "FlaskPlugin"), + "quart": Plugin(".quart_plugin", __name__, "QuartPlugin"), "falcon": Plugin(".falcon_plugin", __name__, "FalconPlugin"), "falcon-asgi": Plugin(".falcon_plugin", __name__, "FalconAsgiPlugin"), "starlette": Plugin(".starlette_plugin", __name__, "StarlettePlugin"), diff --git a/spectree/plugins/quart_plugin.py b/spectree/plugins/quart_plugin.py new file mode 100644 index 00000000..c0a4dab7 --- /dev/null +++ b/spectree/plugins/quart_plugin.py @@ -0,0 +1,290 @@ +import inspect +from typing import Any, Callable, Mapping, Optional, Tuple, get_type_hints + +from pydantic import BaseModel, ValidationError +from quart import Blueprint, abort, current_app, jsonify, make_response, request +from werkzeug.routing import parse_converter_args + +from .._types import ModelType +from ..response import Response +from ..utils import get_multidict_items, werkzeug_parse_rule +from .base import BasePlugin, Context + + +class QuartPlugin(BasePlugin): + blueprint_state = None + FORM_MIMETYPE = ("application/x-www-form-urlencoded", "multipart/form-data") + ASYNC = True + + def find_routes(self): + for rule in current_app.url_map.iter_rules(): + if any( + str(rule).startswith(path) + for path in (f"/{self.config.path}", "/static") + ): + continue + if rule.endpoint.startswith("openapi"): + continue + if ( + self.blueprint_state + and self.blueprint_state.url_prefix + and ( + not str(rule).startswith(self.blueprint_state.url_prefix) + or str(rule).startswith( + "/".join([self.blueprint_state.url_prefix, self.config.path]) + ) + ) + ): + continue + yield rule + + def bypass(self, func, method): + return method in ["HEAD", "OPTIONS"] + + def parse_func(self, route: Any): + if self.blueprint_state: + func = self.blueprint_state.app.view_functions[route.endpoint] + else: + func = current_app.view_functions[route.endpoint] + + # view class: https://flask.palletsprojects.com/en/1.1.x/views/ + if getattr(func, "view_class", None): + cls = getattr(func, "view_class") + for method in route.methods: + view = getattr(cls, method.lower(), None) + if view: + yield method, view + else: + for method in route.methods: + yield method, func + + def parse_path( + self, + route: Optional[Mapping[str, str]], + path_parameter_descriptions: Optional[Mapping[str, str]], + ) -> Tuple[str, list]: + subs = [] + parameters = [] + + for converter, arguments, variable in werkzeug_parse_rule(str(route)): + if converter is None: + subs.append(variable) + continue + subs.append(f"{{{variable}}}") + + args: tuple = () + kwargs: dict = {} + + if arguments: + args, kwargs = parse_converter_args(arguments) + + schema = None + if converter == "any": + schema = { + "type": "string", + "enum": args, + } + elif converter == "int": + schema = { + "type": "integer", + "format": "int32", + } + if "max" in kwargs: + schema["maximum"] = kwargs["max"] + if "min" in kwargs: + schema["minimum"] = kwargs["min"] + elif converter == "float": + schema = { + "type": "number", + "format": "float", + } + elif converter == "uuid": + schema = { + "type": "string", + "format": "uuid", + } + elif converter == "path": + schema = { + "type": "string", + "format": "path", + } + elif converter == "string": + schema = { + "type": "string", + } + for prop in ["length", "maxLength", "minLength"]: + if prop in kwargs: + schema[prop] = kwargs[prop] + elif converter == "default": + schema = {"type": "string"} + + description = ( + path_parameter_descriptions.get(variable, "") + if path_parameter_descriptions + else "" + ) + parameters.append( + { + "name": variable, + "in": "path", + "required": True, + "schema": schema, + "description": description, + } + ) + + return "".join(subs), parameters + + async def request_validation(self, request, query, json, form, headers, cookies): + """ + req_query: werkzeug.datastructures.ImmutableMultiDict + req_json: dict + req_headers: werkzeug.datastructures.EnvironHeaders + req_cookies: werkzeug.datastructures.ImmutableMultiDict + """ + req_query = get_multidict_items(request.args) + req_headers = dict(iter(request.headers)) or {} + req_cookies = get_multidict_items(request.cookies) or {} + has_data = request.method not in ("GET", "DELETE") + use_json = json and has_data and request.mimetype == "application/json" + use_form = ( + form + and has_data + and any([x in request.mimetype for x in self.FORM_MIMETYPE]) + ) + + request.context = Context( + query.parse_obj(req_query) if query else None, + json.parse_obj(await request.get_json(silent=True) or {}) + if use_json + else None, + form.parse_obj(self._fill_form(request)) if use_form else None, + headers.parse_obj(req_headers) if headers else None, + cookies.parse_obj(req_cookies) if cookies else None, + ) + + def _fill_form(self, request) -> dict: + req_data = get_multidict_items(request.form) + req_data.update(get_multidict_items(request.files) if request.files else {}) + return req_data + + async def validate( + self, + func: Callable, + query: Optional[ModelType], + json: Optional[ModelType], + form: Optional[ModelType], + headers: Optional[ModelType], + cookies: Optional[ModelType], + resp: Optional[Response], + before: Callable, + after: Callable, + validation_error_status: int, + skip_validation: bool, + *args: Any, + **kwargs: Any, + ): + response, req_validation_error, resp_validation_error = None, None, None + try: + await self.request_validation(request, query, json, form, headers, cookies) + if self.config.annotations: + annotations = get_type_hints(func) + for name in ("query", "json", "form", "headers", "cookies"): + if annotations.get(name): + kwargs[name] = getattr(request.context, name) + except ValidationError as err: + req_validation_error = err + response = await make_response( + jsonify(err.errors()), validation_error_status + ) + + before(request, response, req_validation_error, None) + if req_validation_error: + after(request, response, req_validation_error, None) + assert response # make mypy happy + abort(response) # type: ignore + + result = ( + await func(*args, **kwargs) + if inspect.iscoroutinefunction(func) + else func(*args, **kwargs) + ) + + status = 200 + rest = [] + if resp and isinstance(result, tuple) and isinstance(result[0], BaseModel): + if len(result) > 1: + model, status, *rest = result + else: + model = result[0] + else: + model = result + + if resp: + expect_model = resp.find_model(status) + if expect_model and isinstance(model, expect_model): + skip_validation = True + result = (model.dict(), status, *rest) + + response = await make_response(result) + + if resp and resp.has_model(): + + model = resp.find_model(response.status_code) + if model and not skip_validation: + try: + model.parse_obj(await response.get_json()) # type: ignore + except ValidationError as err: + resp_validation_error = err + response = await make_response( + jsonify({"message": "response validation error"}), 500 + ) + + after(request, response, resp_validation_error, None) + + return response + + def register_route(self, app): + app.add_url_rule( + self.config.spec_url, + endpoint=f"openapi_{self.config.path}", + view_func=lambda: jsonify(self.spectree.spec), + ) + + if isinstance(app, Blueprint): + + def gen_doc_page(ui): + spec_url = self.config.spec_url + if self.blueprint_state.url_prefix is not None: + spec_url = "/".join( + ( + self.blueprint_state.url_prefix.rstrip("/"), + self.config.spec_url.lstrip("/"), + ) + ) + + return self.config.page_templates[ui].format( + spec_url=spec_url, + spec_path=self.config.path, + **self.config.swagger_oauth2_config(), + ) + + for ui in self.config.page_templates: + app.add_url_rule( + f"/{self.config.path}/{ui}/", + endpoint=f"openapi_{self.config.path}_{ui.replace('.', '_')}", + view_func=lambda ui=ui: gen_doc_page(ui), + ) + + app.record(lambda state: setattr(self, "blueprint_state", state)) + else: + for ui in self.config.page_templates: + app.add_url_rule( + f"/{self.config.path}/{ui}/", + endpoint=f"openapi_{self.config.path}_{ui}", + view_func=lambda ui=ui: self.config.page_templates[ui].format( + spec_url=self.config.spec_url, + spec_path=self.config.path, + **self.config.swagger_oauth2_config(), + ), + ) diff --git a/tests/import_module/test_quart_plugin.py b/tests/import_module/test_quart_plugin.py new file mode 100644 index 00000000..6a63d96a --- /dev/null +++ b/tests/import_module/test_quart_plugin.py @@ -0,0 +1,4 @@ +from spectree import SpecTree + +SpecTree("quart") +print("=> passed quart plugin import test") diff --git a/tests/quart_imports/__init__.py b/tests/quart_imports/__init__.py new file mode 100644 index 00000000..8056bbf8 --- /dev/null +++ b/tests/quart_imports/__init__.py @@ -0,0 +1,17 @@ +from .dry_plugin_quart import ( + test_quart_doc, + test_quart_no_response, + test_quart_return_model, + test_quart_skip_validation, + test_quart_validate, + test_quart_validation_error_response_status_code, +) + +__all__ = [ + "test_quart_return_model", + "test_quart_skip_validation", + "test_quart_validation_error_response_status_code", + "test_quart_doc", + "test_quart_validate", + "test_quart_no_response", +] diff --git a/tests/quart_imports/dry_plugin_quart.py b/tests/quart_imports/dry_plugin_quart.py new file mode 100644 index 00000000..bcf5e008 --- /dev/null +++ b/tests/quart_imports/dry_plugin_quart.py @@ -0,0 +1,157 @@ +import asyncio + +import pytest + + +def test_quart_skip_validation(client): + client.set_cookie("quart", "pub", "abcdefg") + + resp = asyncio.run( + client.post( + "/api/user_skip/quart?order=1", + json=dict(name="quart", limit=10), + headers={"Content-Type": "application/json"}, + ) + ) + resp_json = asyncio.run(resp.json) + assert resp.status_code == 200, resp_json + assert resp.headers.get("X-Validation") is None + assert resp.headers.get("X-API") == "OK" + assert resp_json["name"] == "quart" + assert resp_json["x_score"] == sorted(resp_json["x_score"], reverse=True) + + +def test_quart_return_model(client): + client.set_cookie("quart", "pub", "abcdefg") + + resp = asyncio.run( + client.post( + "/api/user_model/quart?order=1", + json=dict(name="quart", limit=10), + headers={"Content-Type": "application/json"}, + ) + ) + resp_json = asyncio.run(resp.json) + assert resp.status_code == 200, resp_json + assert resp.headers.get("X-Validation") is None + assert resp.headers.get("X-API") == "OK" + assert resp_json["name"] == "quart" + assert resp_json["score"] == sorted(resp_json["score"], reverse=True) + + +@pytest.mark.parametrize( + ["test_client_and_api", "expected_status_code"], + [ + pytest.param( + {"api_kwargs": {}, "endpoint_kwargs": {}}, + 422, + id="default-global-status-without-override", + ), + pytest.param( + {"api_kwargs": {}, "endpoint_kwargs": {"validation_error_status": 400}}, + 400, + id="default-global-status-with-override", + ), + pytest.param( + {"api_kwargs": {"validation_error_status": 418}, "endpoint_kwargs": {}}, + 418, + id="overridden-global-status-without-override", + ), + pytest.param( + { + "api_kwargs": {"validation_error_status": 400}, + "endpoint_kwargs": {"validation_error_status": 418}, + }, + 418, + id="overridden-global-status-with-override", + ), + ], + indirect=["test_client_and_api"], +) +def test_quart_validation_error_response_status_code( + test_client_and_api, expected_status_code +): + app_client, _ = test_client_and_api + + resp = asyncio.run(app_client.get("/ping")) + + assert resp.status_code == expected_status_code + + +@pytest.mark.parametrize( + "test_client_and_api, expected_doc_pages", + [ + pytest.param({}, ["redoc", "swagger"], id="default-page-templates"), + pytest.param( + {"api_kwargs": {"page_templates": {"custom_page": "{spec_url}"}}}, + ["custom_page"], + id="custom-page-templates", + ), + ], + indirect=["test_client_and_api"], +) +def test_quart_doc(test_client_and_api, expected_doc_pages): + client, api = test_client_and_api + + resp = asyncio.run(client.get("/apidoc/openapi.json")) + assert asyncio.run(resp.json) == api.spec + + for doc_page in expected_doc_pages: + resp = asyncio.run(client.get(f"/apidoc/{doc_page}/")) + assert resp.status_code == 200 + + resp = asyncio.run(client.get(f"/apidoc/{doc_page}")) + assert resp.status_code == 308 + + +def test_quart_validate(client): + resp = asyncio.run(client.get("/ping")) + assert resp.status_code == 422 + assert resp.headers.get("X-Error") == "Validation Error" + + resp = asyncio.run(client.get("/ping", headers={"lang": "en-US"})) + resp_json = asyncio.run(resp.json) + assert resp_json == {"msg": "pong"} + assert resp.headers.get("X-Error") is None + assert resp.headers.get("X-Validation") == "Pass" + + resp = asyncio.run(client.post("api/user/quart")) + assert resp.status_code == 422 + assert resp.headers.get("X-Error") == "Validation Error" + + client.set_cookie("quart", "pub", "abcdefg") + for fragment in ("user", "user_annotated"): + resp = asyncio.run( + client.post( + f"/api/{fragment}/quart?order=1", + json=dict(name="quart", limit=10), + headers={"Content-Type": "application/json"}, + ) + ) + resp_json = asyncio.run(resp.json) + assert resp.status_code == 200, resp_json + assert resp.headers.get("X-Validation") is None + assert resp.headers.get("X-API") == "OK" + assert resp_json["name"] == "quart" + assert resp_json["score"] == sorted(resp_json["score"], reverse=True) + + resp = asyncio.run( + client.post( + f"/api/{fragment}/quart?order=0", + json=dict(name="quart", limit=10), + headers={"Content-Type": "application/json"}, + ) + ) + resp_json = asyncio.run(resp.json) + assert resp.status_code == 200, resp_json + assert resp_json["score"] == sorted(resp_json["score"], reverse=False) + + +def test_quart_no_response(client): + resp = asyncio.run(client.get("/api/no_response")) + assert resp.status_code == 200 + + resp = asyncio.run( + client.post("/api/no_response", json={"name": "foo", "limit": 1}) + ) + assert resp.status_code == 200 diff --git a/tests/test_plugin_quart.py b/tests/test_plugin_quart.py new file mode 100644 index 00000000..6272bc7d --- /dev/null +++ b/tests/test_plugin_quart.py @@ -0,0 +1,329 @@ +from random import randint + +import pytest +from quart import Quart, jsonify, request + +from spectree import Response, SpecTree + +from .common import ( + JSON, + SECURITY_SCHEMAS, + Cookies, + Headers, + Order, + Query, + Resp, + StrDict, + api_tag, +) + + +def before_handler(req, resp, err, _): + if err: + resp.headers["X-Error"] = "Validation Error" + + +def after_handler(req, resp, err, _): + resp.headers["X-Validation"] = "Pass" + + +def api_after_handler(req, resp, err, _): + resp.headers["X-API"] = "OK" + + +api = SpecTree("quart", before=before_handler, after=after_handler, annotations=True) +app = Quart(__name__) +app.config["TESTING"] = True + +api_secure = SpecTree("quart", security_schemes=SECURITY_SCHEMAS) +app_secure = Quart(__name__) +app_secure.config["TESTING"] = True + +api_global_secure = SpecTree( + "quart", security_schemes=SECURITY_SCHEMAS, security={"auth_apiKey": []} +) +app_global_secure = Quart(__name__) +app_global_secure.config["TESTING"] = True + + +@app.route("/ping") +@api.validate(headers=Headers, resp=Response(HTTP_200=StrDict), tags=["test", "health"]) +async def ping(): + """summary + + description""" + return jsonify(msg="pong") + + +@app.route("/api/user/", methods=["POST"]) +@api.validate( + query=Query, + json=JSON, + cookies=Cookies, + resp=Response(HTTP_200=Resp, HTTP_401=None), + tags=[api_tag, "test"], + after=api_after_handler, +) +async def user_score(name): + score = [randint(0, request.context.json.limit) for _ in range(5)] + score.sort(reverse=True if request.context.query.order == Order.desc else False) + assert request.context.cookies.pub == "abcdefg" + assert request.cookies["pub"] == "abcdefg" + return jsonify(name=request.context.json.name, score=score) + + +@app.route("/api/user_annotated/", methods=["POST"]) +@api.validate( + resp=Response(HTTP_200=Resp, HTTP_401=None), + tags=[api_tag, "test"], + after=api_after_handler, +) +async def user_score_annotated(name, query: Query, json: JSON, cookies: Cookies): + score = [randint(0, json.limit) for _ in range(5)] + score.sort(reverse=True if query.order == Order.desc else False) + assert cookies.pub == "abcdefg" + assert request.cookies["pub"] == "abcdefg" + return jsonify(name=json.name, score=score) + + +@app.route("/api/user_skip/", methods=["POST"]) +@api.validate( + query=Query, + json=JSON, + cookies=Cookies, + resp=Response(HTTP_200=Resp, HTTP_401=None), + tags=[api_tag, "test"], + after=api_after_handler, + skip_validation=True, +) +async def user_score_skip_validation(name): + score = [randint(0, request.context.json.limit) for _ in range(5)] + score.sort(reverse=True if request.context.query.order == Order.desc else False) + assert request.context.cookies.pub == "abcdefg" + assert request.cookies["pub"] == "abcdefg" + return jsonify(name=request.context.json.name, x_score=score) + + +@app.route("/api/user_model/", methods=["POST"]) +@api.validate( + query=Query, + json=JSON, + cookies=Cookies, + resp=Response(HTTP_200=Resp, HTTP_401=None), + tags=[api_tag, "test"], + after=api_after_handler, +) +async def user_score_model(name): + score = [randint(0, request.context.json.limit) for _ in range(5)] + score.sort(reverse=True if request.context.query.order == Order.desc else False) + assert request.context.cookies.pub == "abcdefg" + assert request.cookies["pub"] == "abcdefg" + return Resp(name=request.context.json.name, score=score), 200 + + +@app.route("/api/user//address/", methods=["GET"]) +@api.validate( + query=Query, + path_parameter_descriptions={ + "name": "The name that uniquely identifies the user.", + "non-existent-param": "description", + }, +) +async def user_address(name, address_id): + return None + + +@app.route("/api/no_response", methods=["GET", "POST"]) +@api.validate( + json=JSON, +) +async def no_response(): + return {} + + +# INFO: ensures that spec is calculated and cached _after_ registering +# view functions for validations. This enables tests to access `api.spec` +# without app_context. +# with app.app_context(): +# api.spec +api.register(app) + + +@pytest.fixture +def client(): + client = app.test_client() + return client + + +@pytest.fixture +def test_client_and_api(request): + api_args = ["quart"] + api_kwargs = {} + endpoint_kwargs = { + "headers": Headers, + "resp": Response(HTTP_200=StrDict), + "tags": ["test", "health"], + } + if hasattr(request, "param"): + api_args.extend(request.param.get("api_args", ())) + api_kwargs.update(request.param.get("api_kwargs", {})) + endpoint_kwargs.update(request.param.get("endpoint_kwargs", {})) + + api = SpecTree(*api_args, **api_kwargs) + app = Quart(__name__) + app.config["TESTING"] = True + + @app.route("/ping") + @api.validate(**endpoint_kwargs) + async def ping(): + """summary + + description""" + return jsonify(msg="pong") + + # INFO: ensures that spec is calculated and cached _after_ registering + # view functions for validations. This enables tests to access `api.spec` + # without app_context. + api.register(app) + test_client = app.test_client() + return test_client, api + + +""" +Secure param check +""" + + +@app_secure.route("/no-secure-ping", methods=["GET"]) +@api_secure.validate( + resp=Response(HTTP_200=StrDict), +) +async def no_secure_ping(): + """ + No auth type is set + """ + return jsonify(msg="pong") + + +@app_secure.route("/apiKey-ping", methods=["GET"]) +@api_secure.validate( + resp=Response(HTTP_200=StrDict), + security={"auth_apiKey": []}, +) +async def apiKey_ping(): + """ + apiKey auth type + """ + return jsonify(msg="pong") + + +@app_secure.route("/apiKey-BasicAuth-ping", methods=["GET"]) +@api_secure.validate( + resp=Response(HTTP_200=StrDict), + security={"auth_apiKey": [], "auth_BasicAuth": []}, +) +async def apiKey_BasicAuth_ping(): + """ + Multiple auth types is set - apiKey and BasicAuth + """ + return jsonify(msg="pong") + + +@app_secure.route("/BasicAuth-ping", methods=["GET"]) +@api_secure.validate( + resp=Response(HTTP_200=StrDict), + security={"auth_BasicAuth": []}, +) +async def BasicAuth_ping(): + """ + BasicAuth auth type + """ + return jsonify(msg="pong") + + +@app_secure.route("/oauth2-flows-ping", methods=["GET"]) +@api_secure.validate( + resp=Response(HTTP_200=StrDict), + security={"auth_oauth2": ["admin", "read"]}, +) +async def oauth_two_ping(): + """ + oauth2 auth type with flow + """ + return jsonify(msg="pong") + + +# with app_secure.app_context(): +# api_secure.spec + +api_secure.register(app_secure) + + +""" +Global secure params check +""" + + +@app_global_secure.route("/global-secure-ping", methods=["GET"]) +@api_global_secure.validate( + resp=Response(HTTP_200=StrDict), +) +async def global_auth_ping(): + """ + global auth type + """ + return jsonify(msg="pong") + + +@app_global_secure.route("/no-secure-override-ping", methods=["GET"]) +@api_global_secure.validate( + security={}, + resp=Response(HTTP_200=StrDict), +) +async def global_no_secure_ping(): + """ + No auth type is set to override + """ + return jsonify(msg="pong") + + +@app_global_secure.route("/oauth2-flows-override-ping", methods=["GET"]) +@api_global_secure.validate( + security={"auth_oauth2": ["admin", "read"]}, + resp=Response(HTTP_200=StrDict), +) +async def global_oauth_two_ping(): + """ + oauth2 auth type with flow to override + """ + return jsonify(msg="pong") + + +@app_global_secure.route("/security_and", methods=["GET"]) +@api_global_secure.validate( + security={"auth_apiKey": [], "auth_apiKey_backup": []}, + resp=Response(HTTP_200=StrDict), +) +async def global_security_and(): + """ + global auth AND + """ + return jsonify(msg="pong") + + +@app_global_secure.route("/security_or", methods=["GET"]) +@api_global_secure.validate( + security=[{"auth_apiKey": []}, {"auth_apiKey_backup": []}], + resp=Response(HTTP_200=StrDict), +) +async def global_security_or(): + """ + global auth OR + """ + return jsonify(msg="pong") + + +# with app_global_secure.app_context(): +# api_global_secure.spec + +api_global_secure.register(app_global_secure)