有没有人最近有问题快速API后台任务?
名称:fastapi 0.99.1
在开始的一年里,我用它,它是非常快的实施。但现在我尝试了一切,它运行起来像一个正常的功能。
我的代码很长,因为我在阅读多个PDF和做解析器。所以我尝试了一些基本的例子来测试核心功能,但它仍然不起作用。
我试过的例子:
-->第一次尝试(测试)
-->第二次尝试(test_blog)-我发现有人在谈论使用blog
- 路由器文件
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)
- 服务文件
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)
正在发生的事情:
1.进入路线
1.只要在运行所有应该在后台运行的任务后返回响应即可。
我所期望的:
1.进入路线
1.返回响应
1.继续在后台运行后台任务功能
编辑1:我创建了一个新的项目,非常基本,只有基本的逻辑和两个功能的工作!所以这不是版本的问题
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"
运行命令:
uvicorn main:app --reload
1条答案
按热度按时间aydmsdu91#
在尝试了edit 1代码后,我发现问题不是关于backgroundTask,而是关于我项目中的其他东西,经过多次尝试,我发现了这个问题!
它在main.py上有一个用于创建日志的自定义中间件,这使得我的backgroundTask像正常函数一样工作,我改变了它,它工作了!