A simple POC that mimics in FastAPI the "g" request lifecycle pseudo global

a_G_for_FastAPI.md

Like Flask "g". Good news you can do the same with FastAPI

"g" is a threadlocal magic object that lets developer add / change / remove attributes during the request lifecycle. Learn more about this "g" here.

There is no OTB equivalent in FastAPI, but thanks to the new contextvars Python 3.7+ module, I made this simple demo.

Any comment or help to improve yhis recipe is welcome.

Hereafter follow the files descriptions and usage for the demo.

requestvars.py

The heart of the stuff. Simple ! Just have a look at the doc of the contextvars module to understand in seconds what it does.

asgi.py

Nothing special here, except the middleware function init_requestvar that stores an empty type.SimpleNamespace object in our context variable at the start of each request cycle.

routes.py

The demo of usage of our FastAPI "g" that is a function (sorry for this, but I dunno how to create a lazily evaluated variable).

Note that no data is passed from the foo_route handler to the double async function.

server.py

Just a simple ordinary uvicorn server for the app that listens on port 8000/

client.py

A simple client that queries in a loop http://localhost:8000?q=xyz and prints the response in an infinite loop. xyz being the first argument of the shell command line that runs client.py.

Run the demo

If you want to run the demo, just create a Python 3.7+ virtual env and :

pip install fastapi uvicorn requests

Open a first terminal and run the server:

python server.py

Open a second terminal and run the client with an argument:

python client.py arg1

It show:

{'result': 'arg1arg1'}
... (same each second)

Open a third terminal and run the client with another argument:

python client.py foo1

It show:

{'result': 'foo1foo1'}
... (same each second)

Repeat the operation on any new terminal you want...

asgi.py

import requestvars
import types
import fastapi
import routes


def create_app() -> fastapi.FastAPI:
    app = fastapi.FastAPI()

    @app.middleware("http")
    async def init_requestvars(request: fastapi.Request, call_next):
        # Customize that SimpleNamespace with hatever you need
        initial_g = types.SimpleNamespace()
        requestvars.request_global.set(initial_g)
        response = await call_next(request)
        return response

    app.include_router(routes.router)
    return app

client.py

import sys
import requests

endpoint = "http://localhost:8000/foo"

query = sys.argv[1]

while True:
    params = {"q": query}
    response = requests.get(endpoint, params=params)
    print(response.json())

requestvars.py

import contextvars
import types
import typing

request_global = contextvars.ContextVar("request_global",
                                        default=types.SimpleNamespace())


# This is the only public API
def g():
    return request_global.get()

routes.py

import asyncio
import fastapi
from requestvars import g

router = fastapi.APIRouter()


@router.get("/foo")
async def foo_route(q: str = ""):
    g().blah = q
    result = await double()
    return {"result": result}


async def double():
    await asyncio.sleep(1.0)
    return g().blah * 2

server.py

import uvicorn
from asgi import create_app


def main():
    app = create_app()
    uvicorn.run(app)


if __name__ == "__main__":
    main()