Example code for lifespan doesn't work, scope issue? #11017
Replies: 4 comments 1 reply
-
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"}
at |
Beta Was this translation helpful? Give feedback.
-
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:
And then
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. |
Beta Was this translation helpful? Give feedback.
-
@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:
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 |
Beta Was this translation helpful? Give feedback.
-
@maxiedaniels 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
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)
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)) |
Beta Was this translation helpful? Give feedback.
-
First Check
Commit to Help
Example Code
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
Beta Was this translation helpful? Give feedback.
All reactions