Friday, September 1, 2023

Python Quart Parse Request + Cheatsheet

import site, os # get quart version with standard libs
sitepackagedirs = [x[0] for x in os.walk(site.getsitepackages()[1])]
print(quart.__name__, [pack for pack in sitedirs if 'quart-' in pack][0][-16:-10])

from quart import Quart, render_template, request

@app.route("/hello")
async def hello():

request.method
request.url
request.headers["X-Bob"]
request.args.get("a") # Query string e.g. example.com/hello?a=2
request.args.getlist()
request.args.to_dict())
await request.get_data() # Full raw body
(await request.form)["name"]
await request.args.getlist()
(await request.get_json())["key"]
request.cookies.get("name")

https://tedboy.github.io/flask/generated/generated/flask.Request.html

The request object is a Request subclass and provides all of the attributes Werkzeug defines plus a few Flask specific ones.

form
    A MultiDict with the parsed form data from POST or PUT requests. Please keep in mind that file uploads will not end up here, but instead in the files attribute.
args
    A MultiDict with the parsed contents of the query string. (The part in the URL after the question mark).
values
    A CombinedMultiDict with the contents of both form and args.
cookies
    A dict with the contents of all cookies transmitted with the request.
stream
    If the incoming form data was not encoded with a known mimetype the data is stored unmodified in this stream for consumption. Most of the time it is a better idea to use data which will give you that data as a string. The stream only returns the data once.
headers
    The incoming request headers as a dictionary like object.
data
    Contains the incoming request data as string in case it came with a mimetype Flask does not handle.
files
    A MultiDict with files uploaded as part of a POST or PUT request. Each file is stored as FileStorage object. It basically behaves like a standard file object you know from Python, with the difference that it also has a save() function that can store the file on the filesystem.
environ
    The underlying WSGI environment.
method
    The current request method (POST, GET etc.)

https://tedboy.github.io/flask/generated/generated/werkzeug.MultiDict.html

to_dict([flat]) Return the contents as regular dict.
getlist(key[, type]) Return the list of items for a given key.


# Requested url: http://192.168.2.100:5000/resultdic?q=Kern&sl=de&tl=ro&dic=IATE&dic=SAP
# http://192.168.2.100:5000/resultdic?q=Kern&sl=de&tl=ro&dic=all

allbidiclist: list[str] = ['Hallo.ro', 'Dict.cc', 'Linguee.com', 'IATE', 'SAP', 'Yandex']

@app.route("/resultdic", methods=["GET"])
async def resultdic():

if request.method == "GET":
searchformterm = request.args.get("q", None)
sourcelang = request.args.get("sl", None)
targetlang = request.args.get("tl", None) # get(key, default=None, type=None)
# Return the default value if the requested data doesn’t exist. If type is provided and is a callable it should convert the value, return it or raise a ValueError if that is not possible.
selecteddiclist = request.args.getlist("dic", None) # list from dic=Dict.cc&dic=SAP
if selecteddiclist: selecteddiclist = alldiclist if selecteddiclist[0] == "all" else selecteddiclist
print("request.args dict:", request.args.to_dict()) # Python dictionary

Cheatsheet

Basic App

from quart import Quart

app = Quart(__name__)

@app.route("/hello")
async def hello():
    return "Hello, World!"

if __name__ == "__main__":
    app.run(debug=True)

Routing

@app.route("/hello/<string:name>")  # example.com/hello/quart
async def hello(name):
    return f"Hello, {name}!"

Request Methods

@app.route("/get")  # GET Only by default
@app.route("/get", methods=["GET", "POST"])  # GET and POST
@app.route("/get", methods=["DELETE"])  # Just DELETE

JSON Responses

@app.route("/hello")
async def hello():
    return {"Hello": "World!"}

Template Rendering

from quart import render_template

@app.route("/hello")
async def hello():
    return await render_template("index.html")  # Required to be in templates/

Configuration

import json
import toml

app.config["VALUE"] = "something"

app.config.from_file("filename.toml", toml.load)
app.config.from_file("filename.json", json.load)

Request

from quart import request

@app.route("/hello")
async def hello():
    request.method
    request.url
    request.headers["X-Bob"]
    request.args.get("a")  # Query string e.g. example.com/hello?a=2
    await request.get_data()  # Full raw body
    (await request.form)["name"]
    (await request.get_json())["key"]
    request.cookies.get("name")

WebSocket

from quart import websocket

@app.websocket("/ws")
async def ws():
    websocket.headers
    while True:
        try:
            data = await websocket.receive()
            await websocket.send(f"Echo {data}")
        except asyncio.CancelledError:
            # Handle disconnect
            raise

Cookies

from quart import make_response

@app.route("/hello")
async def hello():
    response = await make_response("Hello")
    response.set_cookie("name", "value")
    return response

Abort

from quart import abort

@app.route("/hello")
async def hello():
    abort(409)

HTTP/2 & HTTP/3 Server Push

from quart import make_push_promise, url_for

@app.route("/hello")
async def hello():
    await make_push_promise(url_for('static', filename='css/minimal.css'))


Source: https://pgjones.gitlab.io

The docs describe the attributes available on the request object (from flask import request) during a request. In most common cases request.data will be empty because it's used as a fallback:

request.data Contains the incoming request data as string in case it came with a mimetype Flask does not handle.

  • request.args: the key/value pairs in the URL query string
  • request.form: the key/value pairs in the body, from a HTML post form, or JavaScript request that isn't JSON encoded
  • request.files: the files in the body, which Flask keeps separate from form. HTML forms must use enctype=multipart/form-data or files will not be uploaded.
  • request.values: combined args and form, preferring args if keys overlap
  • request.json: parsed JSON data. The request must have the application/json content type, or use request.get_json(force=True) to ignore the content type.

All of these are MultiDict instances (except for json). You can access values using:

  • request.form['name']: use indexing if you know the key exists
  • request.form.get('name'): use get if the key might not exist
  • request.form.getlist('name'): use getlist if the key is sent multiple times and you want a list of values. get only returns the first value.

Run in debug mode:

quart_cfg = hypercorn.Config()
quart_cfg.bind = ["0.0.0.0:8000"]
app = Quart(__name__)
...
async def main():
    await hypercorn.asyncio.serve(app,quart_cfg)

if __name__ == '__main__':
    client.loop.set_debug(True)
    client.loop.run_until_complete(main())

Run function at startup:

a background task that is started on startup, 

async def schedule():
    while True:
        await asyncio.sleep(1)
        await do_work()

@app.before_serving
async def startup():
    app.add_background_task(schedule)

which will run schedule for the lifetime of the app, being cancelled at shutdown.

The ASGI lifespan specification includes the ability for awaiting coroutines before the first byte is received and after the final byte is sent, through the startup and shutdown lifespan events. This is particularly useful for creating and destroying connection pools. Quart supports this via the decorators before_serving() and after_serving(), which function like before_first_request(), and while_serving() which expects a function that returns a generator.

@app.before_serving
async def create_db_pool():
    app.db_pool = await ...
    g.something = something

@app.before_serving
async def use_g():
    g.something.do_something()

@app.while_serving
async def lifespan():
    ...  # startup
    yield
    ...  # shutdown

@app.route("/")
async def index():
    app.db_pool.execute(...)
    # g.something is not available here

@app.after_serving
async def create_db_pool():
    await app.db_pool.close()

AI response: Quart does not have a direct equivalent to Flask's @response.call_on_close. However, Quart provides a way to handle tasks in the background which can be used to achieve similar functionality.

from quart import Quart, request, jsonify  
from quart.ctx import copy_current_websocket_context  
 
app = Quart(__name__)  
 
@app.route('/')  
async def index():  
    @copy_current_websocket_context  
    async def after_request():  
        # Code to be executed after the response is sent  
        pass  
 
    app.add_background_task(after_request)  
    return jsonify({'message': 'Hello, World!'})  

In this example, the after_request function will be executed after the response is sent. The copy_current_websocket_context decorator is used to preserve the context for the function when it's run in the background. Please note that errors raised in a background task are logged but otherwise ignored allowing the app to continue - much like with request/websocket handling errors.

Source: https://stackoverflow.com

at 9:13 PM
Labels: , , ,