Do anyone had problem recently with Fast API BackgroundTask?
Name: fastapi Version: 0.99.1
In the begin of the year I used and it was very fast to implement it. But now I tried everything and it runs like a normal function.
My code it was very long that cause I was reading multiple pdfs and doing the parser. So I tried with some basic example to test the core functionality but it didn't work still.
Examples that I tried:
--> First attempt (test)
--> Second attempt (test_asyncio) - I found someone talking about using asyncio
-router file
from fastapi import APIRouter, BackgroundTasks, Depends, status
from app.config.dbconfig import get_db
from app.services.pdf import ServicePdf
from app.utils.service_result import ServiceResult, handle_result
router = APIRouter(
prefix="/pdf",
tags=["pdf"],
responses={404: {"description": "Not found"}},
)
@router.post("/test", status_code=status.HTTP_201_CREATED)
async def parse_candidate_pdf(
background_tasks: BackgroundTasks,
db: get_db = Depends(),
):
background_tasks.add_task(ServicePdf(db).test, "name")
return handle_result(ServiceResult(None), status.HTTP_201_CREATED)
@router.post("/test-asyncio", status_code=status.HTTP_201_CREATED)
async def parse_candidate_pdf_asyncio(
background_tasks: BackgroundTasks,
db: get_db = Depends(),
):
background_tasks.add_task(ServicePdf(db).test_asyncio, "name")
return handle_result(ServiceResult(None), status.HTTP_201_CREATED)
-service file
import asyncio
import time
from app.services.main import AppService
class ServicePdf(AppService):
def test(self, name):
time.sleep(5)
print("Awake now!", name)
async def test_asyncio(self, name):
await asyncio.sleep(5)
print("Awake now!", name)
What is happening:
- Enter on the route
- Just return the response after run all that have inside of the task that was suppose to be in background.
What I expected:
- Enter on the route
- Return the response
- Continue running the background task function on the background
Edit1: I created a new project, very basic only with the base logic and both functions worked! so it's not about the version
main.py
import time
import asyncio
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
async def process_data_asyncio(data: str):
await asyncio.sleep(5)
print(f"Processed data: {data}")
def process_data(data: str):
time.sleep(5)
print(f"Processed data: {data}")
@app.post("/process-data-asyncio/")
async def parse_asyncio(data: str, background_tasks: BackgroundTasks):
background_tasks.add_task(process_data_asyncio, data)
return {"message": "Data processing started in the background."}
@app.post("/process-data/")
async def parse(data: str, background_tasks: BackgroundTasks):
background_tasks.add_task(process_data, data)
return {"message": "Data processing started in the background."}
@app.get("/")
def home():
return "api"
command to run:
uvicorn main:app --reload
