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()

Looks great!
There is one gap I found when working with middleware.
The content on g() doesn't exist after the request.

    async def test_g(request: fastapi.Request, call_next):
        # Customize that SimpleNamespace with hatever you need
        g().test = 1
        
        response = await call_next(request)
        
        print(g().test) # failed

        return response

Yes, g() is intended to be used in functions / coroutines / route handlers where the request is not (easily) available. And you're not supposed, unless I'm missing something, to control the order of middlewares execution. Instead, you should customize the init_requestvars middleware, putting whatever you want / need in the initial SimpleNamespace() object.

Change the lines 10-16 of asgi.py with this...

    @app.middleware("http")
    async def init_requestvars(request: fastapi.Request, call_next):
        # Customized  SimpleNamespace
        initial_g = types.SimpleNamespace(test=1)
        requestvars.request_global.set(initial_g)
        response = await call_next(request)
        return response

You said
> Any comment on the Gist is welcome
I think you are using it correctly, contextvar is the way to go. Though it might be important to .reset the request_global.
Shameless plug, here's my lib, starlette-context that does exactly that with some additional features.

So I'm here, because like a number of people I wanted some sort of global context object. Personally it was to do a consumer producer paradigm where when the request would come in, put something on a queue, and then I'd have a non blocking task in the background doing stuff.

I tried all sorts of stuff including this which also didn't work for me the way I wanted.

Here's how I did it:

persistent_object = type('', (), {})()
persistent_object.gateways = {}
persistent_object.services = {}

I don't know exactly how this works, but it's python magic that creates a pass by reference object. I use this to store stuff and consume. I don't know if it's safe, but so far I've had no issues

Then I create new threads:

import threading

def dont_block(f):
    def wrap(*args):
        threading.Thread(target=f, args=args, daemon=True).start()
    return wrap

@dont_block
def build_gateways_dictionary():
    while True:
        sleep(1)
        print(persistent_object.gateways)
        
@dont_block
def build_services_dictionary():
    while True:
        sleep(1)
        print(persistent_object.gateways)

That do stuff because trying to attach to the main loop thats already going was a nightmare for me.

Love you dude!!

If you set a property of SimpleNamespace directly like g().blah = q , then the blah be the same in different requests. Instead if you set variables with using set() , then variables will be the different in different requests.

from requestvars import g, request_global

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


# Do like this
@router.get("/foo")
async def foo_route(q: str = ""):
    namespace = request_global.get()
    namespace.blah = q
    request_global.set(namespace)

    result = await double()
    return {"result": result}