🤔 What is correct way to do DB operation after successfully streaming response?

[jd-solanki]

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

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    chunks = []
    for i in range(10):
        chunks.append(i)
        yield b"some fake video bytes"
    print("Streaming Done!", chunks)
    # 🤔 How can I perform DB operation here?


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer())

Description

Hi 👋🏻

I'm streaming the LLM response and once it's completed I want to store the generated chunk (chunks in the above example) in the database.

I can pass DB dependency in fake_video_streamer but I think that is not the correct way because once the streaming is done DB session (I'm not sure about session or connection but something of DB) might get closed and we won't able to perform the insert operation.

Hence, What is the best way to do post-streaming insertion in DB?

Thanks 😇

Operating System

Linux

Operating System Details

N/A

FastAPI Version

0.110.1

Pydantic Version

2.6.4

Python Version

Python 3.12.2

Additional Context

No response

[YuriiMotov]

It seems that you should use the same approach as for background tasks (https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/#background-tasks-and-dependencies-with-yield-technical-details)

create a new database session inside of the background task

So, you need to pass session_maker to your generator function (use dependency to get it in your endpoint function) and create session inside your generator function when you need to interact with DB.

But I think this topic is a bit badly described in the documentation and should be improved.

Also, it seems that this statement from the documentation is a bit confusing:

The code following the yield statement is executed after the response has been delivered

(https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/#background-tasks-and-dependencies-with-yield-technical-details)

It's true for usual responses, but it's false for StreamingResponse:

INFO:     Started server process [1009599]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
before yield
after yield
INFO:     127.0.0.1:56070 - "GET / HTTP/1.1" 200 OK
Streaming Done! [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

[jd-solanki]

Thanks that was a great explanation. You must be an expert here. Thanks 😇

If the background task runs even before the streaming response completes then we won't be able to get all data that is streamed. Hence, We'll have different data in DB than streamed data.

Do you think I should raise feature request if there's no way to execute DB operation after StreamingResponse finishes?

[jd-solanki]

Thanks to Doctor from discord I was able to finalize this:

from time import sleep
from typing import Any

from fastapi import BackgroundTasks, FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


def save_to_db(data: Any):
    print("Saving to the database:")
    print(data)


def fake_streamer(background_tasks: BackgroundTasks):
    chunks = []
    for i in range(3):
        sleep(1)
        chunks.append(i)
        yield f"Streaming: {i}\n"
    print("Streaming Done!", chunks)
    background_tasks.add_task(save_to_db, chunks)


@app.get("/")
async def main(background_tasks: BackgroundTasks):
    return StreamingResponse(fake_streamer(background_tasks))

We're still discussing what'll be performant way to provide DB in here.

[hamdan-27]

Hi @jd-solanki, I have the same question and am wondering if you found the optimal way of performing DB tasks?