Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_blueprint_url_prefix(app: Pint, req_ctx) -> None:
blueprint = Blueprint('blueprint', __name__)
prefix = Blueprint('prefix', __name__, url_prefix='/prefix')
@app.route('/page/')
@blueprint.route('/page/')
@prefix.route('/page/')
async def route() -> ResponseReturnValue:
return 'OK'
app.register_blueprint(blueprint, url_prefix='/blueprint')
app.register_blueprint(prefix)
async with req_ctx(app, '/', method='GET'):
assert '/page/' == url_for('route')
assert '/prefix/page/' == url_for('prefix.route')
assert '/blueprint/page/' == url_for('blueprint.route')
async with req_ctx(app, '/page/', method='GET'):
async def test_blueprint_method_view(app: Pint) -> None:
blueprint = Blueprint('blueprint', __name__)
class Views(MethodView):
async def get(self) -> ResponseReturnValue:
return 'GET'
async def post(self) -> ResponseReturnValue:
return 'POST'
blueprint.add_url_rule('/', view_func=Views.as_view('simple'))
app.register_blueprint(blueprint)
test_client = app.test_client()
response = await test_client.get('/')
assert 'GET' == (await response.get_data(raw=False))
response = await test_client.post('/')
assert 'POST' == (await response.get_data(raw=False))
async def test_blueprint_url_prefix() -> None:
app = Quart(__name__)
blueprint = Blueprint("blueprint", __name__)
prefix = Blueprint("prefix", __name__, url_prefix="/prefix")
@app.route("/page/")
@blueprint.route("/page/")
@prefix.route("/page/")
async def route() -> ResponseReturnValue:
return "OK"
app.register_blueprint(blueprint, url_prefix="/blueprint")
app.register_blueprint(prefix)
async with app.test_request_context("/page/"):
assert request.blueprint is None
async with app.test_request_context("/prefix/page/"):
assert request.blueprint == "prefix"
async def test_url_for_blueprint_relative(app: Quart) -> None:
blueprint = Blueprint("blueprint", __name__)
@blueprint.route("/")
def index() -> str:
return ""
app.register_blueprint(blueprint, url_prefix="/blue")
async with app.test_request_context("/blue/"):
assert url_for(".index") == "/blue/"
assert url_for("index") == "/"
async def test_blueprint_method_view() -> None:
app = Quart(__name__)
blueprint = Blueprint("blueprint", __name__)
class Views(MethodView):
async def get(self) -> ResponseReturnValue:
return "GET"
async def post(self) -> ResponseReturnValue:
return "POST"
blueprint.add_url_rule("/", view_func=Views.as_view("simple"))
app.register_blueprint(blueprint)
test_client = app.test_client()
response = await test_client.get("/")
assert "GET" == (await response.get_data(raw=False))
response = await test_client.post("/")
async def test_blueprint_websocket() -> None:
app = Quart(__name__)
blueprint = Blueprint("blueprint", __name__)
@blueprint.websocket("/ws/")
async def ws() -> None:
while True:
await websocket.send(websocket.blueprint.encode())
app.register_blueprint(blueprint)
test_client = app.test_client()
async with test_client.websocket("/ws/") as test_websocket:
result = await test_websocket.receive()
assert cast(bytes, result) == b"blueprint"
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import yaml
from apispec import APISpec, yaml_utils
from quart import Blueprint, request, jsonify, render_template, current_app
from zmq.asyncio import Socket
from flowapi import __version__
blueprint = Blueprint("spec", __name__)
async def get_spec(socket: Socket, request_id: str) -> APISpec:
"""
Construct open api spec by interrogating FlowMachine.
Parameters
----------
socket : Socket
request_id : str
Unique id of the request
Returns
-------
APISpec
The specification object
from quart import Blueprint, jsonify, request
from typing import Any, Dict
nodes_controller = Blueprint('nodes', __name__)
@nodes_controller.route('/', methods=['GET'])
async def list_of_nodes() -> Any:
class NodeGetAllEntryResponse(object):
def __init__(self, name: str, id: int, health: str, mode: str, uptime: int, vendor_code: int) -> None:
self.name = name
self.id = id
self.health = health
self.mode = mode
self.uptime = uptime
self.vendor_code = vendor_code
def serialise(self) -> Dict[str, Any]:
return {
'name': self.name,
app = Pint('Custom', no_openapi=True)
@app.route('/openapi.json')
# add other decorators if desired
async def openapi():
# add other logic if desired
return jsonify(app.__schema__)
"""
super().__init__(*args, **kwargs)
self.config['JSON_SORT_KEYS'] = False
if not no_openapi:
self.add_url_rule('/openapi.json', 'openapi',
OpenApiView.as_view('openapi', self), ['GET', 'OPTIONS'])
class PintBlueprint(BaseRest, Blueprint):
"""Use this instead of :class:`quart.Blueprint` objects to allow using Resource class objects with them"""
def __init__(self, *args, **kwargs) -> None:
"""Will forward all arguments and keyword arguments to Blueprints"""
super().__init__(*args, **kwargs)
def register(self, app: Pint, first_registration: bool, *, url_prefix: Optional[str] = None) -> None:
"""override the base :meth:`~quart.Blueprint.register` method to add the resources to the app registering
this blueprint, then call the parent register method
"""
prefix = url_prefix or self.url_prefix or ''
app.resources.extend([(res, f'{prefix}{path}', methods) for res, path, methods in self._resources])
super().register(app, first_registration, url_prefix=url_prefix)
class OpenApiView(Resource):
"""The :class:`Resource` used for the '/openapi.json' route
from quart import abort, Blueprint, current_app, jsonify, request
blueprint = Blueprint('reviews', __name__)
@blueprint.route('/reviews/', methods=['POST'])
async def add_review():
data = await request.get_json()
film_id = data['film_id']
rating = int(data['rating'])
async with current_app.pool.acquire() as connection:
await connection.execute(
"""INSERT INTO review (film_id, rating)
VALUES ($1, $2)""",
film_id, rating,
)
return jsonify({
'film_id': film_id,
'rating': rating,