flask.g for FastAPI.

fastapi_globals.py

"""
This allows to use global variables inside the FastAPI application using async mode.

# Usage

Just import `g` and then access (set/get) attributes of it:
```python
from your_project.globals import g


g.foo = "foo"

# In some other code
assert g.foo == "foo"
```

Best way to utilize the global `g` in your code is to set the desired
value in a FastAPI dependency, like so:
```python
async def set_global_foo() -> None:
    g.foo = "foo"


@app.get("/test/", dependencies=[Depends(set_global_foo)])
async def test():
    assert g.foo == "foo"
```

# Setup

Add the `GlobalsMiddleware` to your app:
```python
app = fastapi.FastAPI(
    title="Your app API",
)
app.add_middleware(GlobalsMiddleware)  # <-- This line is necessary
```

Then just use it. ;-)

# Default values

You may use `g.set_default("name", some_value)` to set a default value
for a global variable. This default value will then be used instead of `None`
when the variable is accessed before it was set.

Note that default values should only be set at startup time, never
inside dependencies or similar. Otherwise you may run into the issue that
the value was already used any thus have a value of `None` set already, which
would result in the default value not being used.
"""
from collections.abc import Awaitable, Callable
from contextvars import ContextVar, copy_context
from typing import Any

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp


class Globals:
    __slots__ = ("_vars", "_defaults")

    _vars: dict[str, ContextVar]
    _defaults: dict[str, Any]

    def __init__(self) -> None:
        object.__setattr__(self, '_vars', {})
        object.__setattr__(self, '_defaults', {})

    def set_default(self, name: str, default: Any) -> None:
        """Set a default value for a variable."""

        # Ignore if default is already set and is the same value
        if (
            name in self._defaults
            and default is self._defaults[name]
        ):
            return

        # Ensure we don't have a value set already - the default will have
        # no effect then
        if name in self._vars:
            raise RuntimeError(
                f"Cannot set default as variable {name} was already set",
            )

        # Set the default already!
        self._defaults[name] = default

    def _get_default_value(self, name: str) -> Any:
        """Get the default value for a variable."""

        default = self._defaults.get(name, None)

        return default() if callable(default) else default

    def _ensure_var(self, name: str) -> None:
        """Ensure a ContextVar exists for a variable."""

        if name not in self._vars:
            default = self._get_default_value(name)
            self._vars[name] = ContextVar(f"globals:{name}", default=default)

    def __getattr__(self, name: str) -> Any:
        """Get the value of a variable."""

        self._ensure_var(name)
        return self._vars[name].get()

    def __setattr__(self, name: str, value: Any) -> None:
        """Set the value of a variable."""

        self._ensure_var(name)
        self._vars[name].set(value)


async def globals_middleware_dispatch(
    request: Request,
    call_next: Callable,
) -> Response:
    """Dispatch the request in a new context to allow globals to be used."""

    ctx = copy_context()

    def _call_next() -> Awaitable[Response]:
        return call_next(request)

    return await ctx.run(_call_next)


class GlobalsMiddleware(BaseHTTPMiddleware):  # noqa
    """Middleware to setup the globals context using globals_middleware_dispatch()."""

    def __init__(self, app: ASGIApp) -> None:
        super().__init__(app, globals_middleware_dispatch)


g = Globals()

Usage:

• Add the GlobalsMiddleware to your app (app.add_middleware(GlobalsMiddleware))
• Import g and use it by accessing attributes of it (like: g.bla = "foo")

It will create a ContextVar object for each attribute you use and allow you global access to that.

I am confused on what you mean by "import g" apologies I am just beginning to dig into middleware.
do you mean fastapi_globals import g is that itself is a singleton?

If you put this into your_app as fastapi_globals.py you might want to do a from your_app.fastapi_globals import g. After this you can just use g.some_var_name to set/fetch global variables.

Thank you for this code snippet, this is very helpful!
I have a few questions in terms of the usage to store values from request within context:

1. How/where is the value being set from the request in the contextVar? Do I need to explicitly call the set_attr to set the values for the fields?
2. How is multiple fields within the same request stored and accessed? I was trying to create a dic of context vars and access them with the request field as key (eg. _vars[REQUEST_ID]). Is that the right way?

1. How/where is the value being set from the request in the contextVar? Do I need to explicitly call the set_attr to set the values for the fields?

I currently use it like this:

async def set_global_something(the_args_it_needs):
    g.something = ...


@app.get(
    "/test/",
    dependencies=[Depends(set_global_something)]  # <-- THIS IS IMPORTANT
)
async def test():
    # Do something with g.something
    pass

Normally the dependency used is something like fetching the user data and then put all of this into g.user or similar. Most of the time those dependencies are added on a router level for a whole block of endpoints.

2. How is multiple fields within the same request stored and accessed? I was trying to create a dic of context vars and access them with the request field as key (eg. _vars[REQUEST_ID]). Is that the right way?

You can just set g.var1, g.var2, g.var3, .... ;-)

RuntimeError: <Token used var=<ContextVar name='globals:db' default=None at 0x7fdd9c684b30> at 0x7fdd9c69b3c0> has already been used once

RuntimeError: <Token used var=<ContextVar name='globals:db' default=None at 0x7fdd9c684b30> at 0x7fdd9c69b3c0> has already been used once

@yxlwfds I did update the gist to our current version. Maybe this fixes your issue?

Hi,

Is it necessary to use async to make this work?

Thanks @ddanier for this gist at the beginning I did not understand but after some testing I was able. For others looking at how to use it, you can check how I use this code to load ml models on fastapi here

Also, I think it could be a good idea to add a cleanup method like this to release memory.

    def cleanup(self):
        """Clear all variables and free memory."""

        self._vars.clear()
        self._defaults.clear()
        del self._vars
        del self._defaults

Should I import GlobalsMiddleware?

@ixjosemi

Hi,

Is it necessary to use async to make this work?

Yes ;-)

@p1utoze

Should I import GlobalsMiddleware?

See docs at the beginning of the file, you also need to add the middleware:
app.add_middleware(GlobalsMiddleware)

@jonra1993

Also, I think it could be a good idea to add a cleanup method like this to release memory.

The idea is more to have a set of globals you always reuse. Those will get "cleaned" up when the request is done, as the ctx.run call is done then (see middleware). Still the idea is to have a predefined set of globals.

For example we tend to use something like g.user for the currently authenticated user.

Does this help to clarify things?

Hey @ddanier , this middleware seems pretty useful for users coming into FastAPI from Flask. Have you considered wrapping it up as a utility library?

Having some issues with this middleware with later versions of fast api referenced here with call_next(request) resulting in an error: tiangolo/fastapi#8187 (reply in thread)

Is there a way to access the g.request_id after the response has been received in another middleware for ex:

from fastapi_globals import GlobalsMiddleware, g

app = FastAPI(lifespan=lifespan)
app.add_middleware(GlobalsMiddleware)


@app.middleware("http")
async def add_request_id(request: Request, call_next):
    g.request_id = str(uuid4())
    logger.info({"event": "Request Start", "request_id": g.request_id})
    response = await call_next(request)
    return response


@app.middleware("http")
async def log_processed_time(request: Request, call_next):
    start_time = time()
    response = await call_next(request)
    process_time = time() - start_time
    logger.info(
        {
            "event": "Request Processed",
            "request_id": g.request_id,
            "process_time": process_time,
        }
    )
    return response

In the above code g.request_id does NOT have the assigned value and instead returns None.

PS: I do NOT want to use any external libraries for the request_id such as asgi-correlation-id