Example code for lifespan doesn't work, scope issue?

[maxiedaniels]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

jobs_container = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    jobs_container_name = "jobs"
    jobs_container = database.create_container_if_not_exists(
        id=jobs_container_name,
        partition_key=PartitionKey(path="/user_id"),
    )
    yield
    ...

app = FastAPI(lifespan=lifespan)

@app.get("/job/status")
def get_job_status(job_id: str = None, current_user: str = Depends(get_current_user)):
    job_user_id = current_user.get("sub")
    try:
        # Query the database for the job
        query = "SELECT * FROM c WHERE c.id = @job_id"
        parameters = [{"name": "@job_id", "value": job_id}, {"name": "@job_user_id", "value": job_user_id}]
        job_items = list(jobs_container.query_items(query=query, parameters=parameters, partition_key=job_user_id))
    ...

Description

According to the docs page for using lifespan, you create an empty variable, and then modify it in the lifespan function. I'm having trouble because when I do that (see code), I get an error that jobs_container is None. If I remove jobs_container = None, the code works, but then my VSCode linters (Pylance and Ruff) freak out and say that jobs_container isn't defined.

Am i doing something wrong here? The docs clearly show setting a variable before lifespan, so why isn't that working in my case?

Operating System

macOS

Operating System Details

No response

FastAPI Version

0.109.0

Pydantic Version

1.10.6

Python Version

3.9.15

Additional Context

No response

[evanmays]

Similar issue. M3 mac

from fastapi import Depends, FastAPI, Header, HTTPException
import uvicorn
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("on")
    yield
    print("off")


app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

uvicorn banana:app --reload --lifespan=on

at site-packages/fastapi/applications.py line 206, in build_middleware_stack
ValueError: too many values to unpack (expected 2)

[patrick91]

this code is working fine for me:

from fastapi import Depends, FastAPI, Header, HTTPException
import uvicorn
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("on")
    yield
    print("off")


app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    return {"message": "Hello World"}

Output:

❯ pdm run uvicorn main:app --reload --lifespan=on
INFO:     Will watch for changes in these directories: ['/private/var/folders/tz/d3_033052gg7k7lrj34yj0lr0000gn/T/tmp.ODxAFQwCVr']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [56411] using StatReload
INFO:     Started server process [56413]
INFO:     Waiting for application startup.
on
INFO:     Application startup complete.
INFO:     127.0.0.1:57983 - "GET / HTTP/1.1" 200 OK
^CINFO:     Shutting down
INFO:     Waiting for application shutdown.
off
INFO:     Application shutdown complete.
INFO:     Finished server process [56413]
INFO:     Stopping reloader process [56411]

Could you run a pip freeze to see if you have any old dependency? 😊

[stefanmachura]

I had the same problem, and I think I came up with the solution just now.

It looks like the problem lies in the fact that in the sample code they are modifying a dict, and @maxiedaniels you are reinitialising a variable.

Try doing it this way:

jobs_container = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    jobs_container_name = "jobs"
    jobs_container["jobs"] = database.create_container_if_not_exists(
        id=jobs_container_name,
        partition_key=PartitionKey(path="/user_id"),
    )
    yield
    ...

And then

job_items = list(jobs_container["jobs"].query_items

I suspect it has something to do with mutating existing variables. If I am right, it would be awesome if the docs could get updated.

[patrick91]

@maxiedaniels I think the main issue is that the variable is a global one, by default python doesn't allow you to change variables declared outside a function.

You have 3 options here:

1. use global jobs_container
2. as @stefanmachura suggested, use a dict (so you change the content of the variable)
3. use state

Here's an example with all 3:

from fastapi import FastAPI
from fastapi import Request
from contextlib import asynccontextmanager


jobs_container = None
jobs = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    global jobs_container
    print("on")
    jobs_container = "container"
    jobs["job1"] = "job1"

    yield {"jobs": {"job2": "job2"}}
    print("off")


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def root(request: Request):
    return {
        "jobs_container": jobs_container,
        "jobs": jobs,
        "jobs_in_state": request.state.jobs,
    }

State is documented here: https://www.starlette.io/lifespan/#lifespan-state

[yawoayite]

@maxiedaniels
Hi I had a similar problem while following the example given in the docs. I was trying to use rabbitmq connection I delcared in the lifespan in one of my endpoints and faced the same problem you described.

I came accross this github project https://github.com/Krishna-D-K/FastAPI_Mongo/tree/main

Instead of using a global variable, he added it to FastAPI in the lifespan and retrieved using Request object.

I will give two examples the first is the code that wasn't working and the second the one working based on the code I read in the linked project

1. My first implementation

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Body
from typing import Any

import os

connection = Any

@asynccontextmanager
async def lifespan(app: FastAPI):
    credentials = pika.PlainCredentials(
        username=os.getenv("RABBITMQ_USERNAME"), password=os.getenv("RABBITMQ_PASS")
    )
    parameters = pika.ConnectionParameters(
        host=os.getenv("RABBITMQ_HOST"),
        credentials=credentials,
        heartbeat=600,
        blocked_connection_timeout=300,
    )
    connection = pika.BlockingConnection(parameters)

    yield {"connection":connection}
    connection.close()

app = FastAPI(lifespan=lifespan)

@app.post("/test")
async def test(request:Request):
    channel = connection.channel()
    queue = channel.queue_declare(queue="trial-queue", durable=True)

2. The working implementation

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Body
from typing import Any

import os


@asynccontextmanager
async def lifespan(app: FastAPI):
    credentials = pika.PlainCredentials(
        username=os.getenv("RABBITMQ_USERNAME"), password=os.getenv("RABBITMQ_PASS")
    )
    parameters = pika.ConnectionParameters(
        host=os.getenv("RABBITMQ_HOST"),
        credentials=credentials,
        heartbeat=600,
        blocked_connection_timeout=300,
    )
    connection = pika.BlockingConnection(parameters)
    
###Adding our variable to FastAPI 
    app.connection = connection

    yield {"connection":connection}
    connection.close()

app = FastAPI(lifespan=lifespan)

@app.post("/test")
async def test(request:Request):
###Here we use the request object to retrieve the variable
    connection = request.app.connection

    channel = connection.channel()
    queue = channel.queue_declare(queue="trial-queue", durable=True)

I suppose if you do your implementation it should work

Could you try it like this ?

from fastapi import FastAPI, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
    jobs_container_name = "jobs"
    jobs_container = database.create_container_if_not_exists(
        id=jobs_container_name,
        partition_key=PartitionKey(path="/user_id"),
    )
  ###Here
    app.jobs_container = jobs_container

    yield


app = FastAPI(lifespan=lifespan)

@app.get("/job/status")
def get_job_status(request:Request,job_id: str = None, current_user: str = Depends(get_current_user)):

###Here
   jobs_container = request.app.jobs_container

    job_user_id = current_user.get("sub")
    try:
        # Query the database for the job
        query = "SELECT * FROM c WHERE c.id = @job_id"
        parameters = [{"name": "@job_id", "value": job_id}, {"name": "@job_user_id", "value": job_user_id}]
        job_items = list(jobs_container.query_items(query=query, parameters=parameters, partition_key=job_user_id))