Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _app() -> Quart:
app = Quart(__name__)
@app.route("/", methods=["GET", "POST"])
def index() -> ResponseReturnValue:
return request.method
@app.route("/gen")
def gen() -> ResponseReturnValue:
def _gen() -> Generator[bytes, None, None]:
yield b"%d" % threading.current_thread().ident
for _ in range(2):
yield b"b"
return _gen(), 200
return app
def serving_app() -> Quart:
app = Quart(__name__)
@app.route('/')
async def index() -> ResponseReturnValue:
return BASIC_DATA, 202, {'X-Test': 'Test'}
@app.route('/push')
async def push() -> ResponseReturnValue:
response = await make_response(BASIC_DATA, 202, {'X-Test': 'Test'})
response.push_promises.add('/')
return response
return app
import ssl
from quart import Quart
app = Quart(__name__)
@app.route('/')
async def index():
return 'Hello'
if __name__ == '__main__':
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.options |= ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1 | ssl.OP_NO_COMPRESSION
ssl_context.set_ciphers('ECDHE+AESGCM')
ssl_context.load_cert_chain(certfile='cert.pem', keyfile='key.pem')
ssl_context.set_alpn_protocols(['h2', 'http/1.1'])
app.run(port=5000, ssl=ssl_context)
from collections import namedtuple
from io import BytesIO
from pathlib import Path
import click
from PIL import Image
from quart import (
make_push_promise, make_response, Quart, render_template, url_for, abort
)
app = Quart(__name__)
Point = namedtuple('Point', ('x', 'y'))
class TileOutOfBoundsError(Exception):
pass
def get_tile(tile_number):
"""
Returns a crop of `img` based on a sequence number `tile_number`.
:param int tile_number: Number of the tile between 0 and `max_tiles`^2.
:raises TileOutOfBoundsError: When `tile_number` exceeds `max_tiles`^2
:rtype PIL.Image:
"""
}
for param in inputs:
expect.append(param)
return self.doc(**params)
@staticmethod
def default_id(resource: str, method: str) -> str:
"""function for creating a default operation id from a resource and method
:param resource: name of the resource endpoint
:param method: the HTTP verb
:return: the id converting camel case to snake_case
"""
return '{0}_{1}'.format(method, camel_to_snake(resource))
class Pint(BaseRest, Quart):
"""Use this instead of instantiating :class:`quart.Quart`
This takes the place of instantiating a :class:`quart.Quart` instance. It will forward
any init arguments to Quart and takes arguments to fill out the metadata for the openapi
documentation it generates that will automatically be accessible via the '/openapi.json'
route.
"""
def __init__(self, *args, no_openapi: bool = False, **kwargs) -> None:
r"""Construct the Pint object, see :meth:`BaseRest.__init__` for an explanation of the args and kwargs.
:param \*args: non-keyword args passed to :class:`BaseRest`
:param no_openapi: set this to True to disable the auto creation of the /openapi.json route
in order to allow custom manipulation of the route
:param \*\*kwargs: keyword args passed to :class:`BaseRest`
from quart import (
abort, jsonify, make_push_promise, Quart, render_template, request, url_for,
)
app = Quart(__name__)
@app.route('/')
async def index():
await make_push_promise(url_for('static', filename='http2.css'))
await make_push_promise(url_for('static', filename='http2.js'))
return await render_template('index.html')
@app.route('/', methods=['POST'])
async def calculate():
data = await request.get_json()
operator = data['operator']
try:
a = int(data['a'])
b = int(data['b'])
import os, requests, json, time, subprocess
from quart import Quart, render_template, redirect, url_for, request
from common import common_setup, common_respond, common_get_verinym, common_reset
from sovrin.credentials import receive_credential_offer, request_credential, store_credential
from sovrin.proofs import create_proof_of_credential
from fetch.agents import offer_service
app = Quart(__name__)
debug = False # Do not enable in production
host = '0.0.0.0'
# In production everyone runs on same port, use 2 here for same-machine testing
port = 5002
issuer_port = 5001
verifier_port = 5003
# We use globals for our server-side session since this is not supported in Quart yet.
prover = {}
request_ip = anchor_ip = received_data = multiple_onboard = service_published = False
pool_handle = 1
stored_credentials = []
@app.route('/')
def create_app():
app = Quart(__name__)
app.config['JSONIFY_PRETTYPRINT_REGULAR'] = False
@app.before_first_request
async def create_db():
dsn = 'postgres://dvdrental:dvdrental@0.0.0.0:5432/dvdrental'
app.pool = await asyncpg.create_pool(dsn, max_size=20) #os.environ['DB_DSN'])
app.register_blueprint(films_blueprint)
app.register_blueprint(reviews_blueprint)
return app
<input type="submit">
'''
# Session name, API ID and hash to use; loaded from environmental variables
SESSION = os.environ.get('TG_SESSION', 'quart')
API_ID = int(get_env('TG_API_ID', 'Enter your API ID: '))
API_HASH = get_env('TG_API_HASH', 'Enter your API hash: ')
# Telethon client
client = TelegramClient(SESSION, API_ID, API_HASH)
client.parse_mode = 'html' # <- Render things nicely
phone = None
# Quart app
app = Quart(__name__)
app.secret_key = 'CHANGE THIS TO SOMETHING SECRET'
# Helper method to format messages nicely
async def format_message(message):
if message.photo:
content = '<img alt="{}" src="data:image/png;base64,{}">'.format(
base64.b64encode(await message.download_media(bytes)).decode(),
message.raw_text
)
else:
# client.parse_mode = 'html', so bold etc. will work!
content = (message.text or '(action message)').replace('\n', '<br>')
return '<p><strong>{}</strong>: {}<sub>{}</sub></p>'.format(
utils.get_display_name(message.sender),
import os
import traceback
from quart import Quart, jsonify
from quart_cors import cors
from quart.exceptions import NotFound
from csvapi.errors import APIError
from csvapi.tableview import TableView
from csvapi.exportview import ExportView
from csvapi.uploadview import UploadView
from csvapi.parseview import ParseView
from csvapi.security import filter_referrers
app = Quart(__name__)
app = cors(app, allow_origin='*')
app.add_url_rule('/api/', view_func=TableView.as_view('table'))
app.add_url_rule('/api//export', view_func=ExportView.as_view('export'))
app.add_url_rule('/apify', view_func=ParseView.as_view('parse'))
app.add_url_rule('/upload', view_func=UploadView.as_view('upload'))
app.before_request(filter_referrers)
conffile = os.environ.get('CSVAPI_CONFIG_FILE') or '../config.py'
app.config.from_pyfile(conffile)
if app.config.get('SENTRY_DSN'):
from raven import Client
app.extensions['sentry'] = Client(app.config['SENTRY_DSN'])