Python Fast API BackgroundTask不工作

0kjbasz6  于 2023-10-14  发布在  Python
关注(0)|答案(1)|浏览(251)

有没有人最近有问题快速API后台任务?
名称:fastapi 0.99.1
在开始的一年里,我用它,它是非常快的实施。但现在我尝试了一切,它运行起来像一个正常的功能。
我的代码很长,因为我在阅读多个PDF和做解析器。所以我尝试了一些基本的例子来测试核心功能,但它仍然不起作用。
我试过的例子:
-->第一次尝试(测试)
-->第二次尝试(test_blog)-我发现有人在谈论使用blog

  • 路由器文件
  1. from fastapi import APIRouter, BackgroundTasks, Depends, status
  2. from app.config.dbconfig import get_db
  3. from app.services.pdf import ServicePdf
  4. from app.utils.service_result import ServiceResult, handle_result
  5. router = APIRouter(
  6. prefix="/pdf",
  7. tags=["pdf"],
  8. responses={404: {"description": "Not found"}},
  9. )
  10. @router.post("/test", status_code=status.HTTP_201_CREATED)
  11. async def parse_candidate_pdf(
  12. background_tasks: BackgroundTasks,
  13. db: get_db = Depends(),
  14. ):
  15. background_tasks.add_task(ServicePdf(db).test, "name")
  16. return handle_result(ServiceResult(None), status.HTTP_201_CREATED)
  17. @router.post("/test-asyncio", status_code=status.HTTP_201_CREATED)
  18. async def parse_candidate_pdf_asyncio(
  19. background_tasks: BackgroundTasks,
  20. db: get_db = Depends(),
  21. ):
  22. background_tasks.add_task(ServicePdf(db).test_asyncio, "name")
  23. return handle_result(ServiceResult(None), status.HTTP_201_CREATED)
  • 服务文件
  1. import asyncio
  2. import time
  3. from app.services.main import AppService
  4. class ServicePdf(AppService):
  5. def test(self, name):
  6. time.sleep(5)
  7. print("Awake now!", name)
  8. async def test_asyncio(self, name):
  9. await asyncio.sleep(5)
  10. print("Awake now!", name)

正在发生的事情:
1.进入路线
1.只要在运行所有应该在后台运行的任务后返回响应即可。
我所期望的:
1.进入路线
1.返回响应
1.继续在后台运行后台任务功能
编辑1:我创建了一个新的项目,非常基本,只有基本的逻辑和两个功能的工作!所以这不是版本的问题
main.py

  1. import time
  2. import asyncio
  3. from fastapi import FastAPI, BackgroundTasks
  4. app = FastAPI()
  5. async def process_data_asyncio(data: str):
  6. await asyncio.sleep(5)
  7. print(f"Processed data: {data}")
  8. def process_data(data: str):
  9. time.sleep(5)
  10. print(f"Processed data: {data}")
  11. @app.post("/process-data-asyncio/")
  12. async def parse_asyncio(data: str, background_tasks: BackgroundTasks):
  13. background_tasks.add_task(process_data_asyncio, data)
  14. return {"message": "Data processing started in the background."}
  15. @app.post("/process-data/")
  16. async def parse(data: str, background_tasks: BackgroundTasks):
  17. background_tasks.add_task(process_data, data)
  18. return {"message": "Data processing started in the background."}
  19. @app.get("/")
  20. def home():
  21. return "api"

运行命令:

  1. uvicorn main:app --reload
aydmsdu9

aydmsdu91#

在尝试了edit 1代码后,我发现问题不是关于backgroundTask,而是关于我项目中的其他东西,经过多次尝试,我发现了这个问题!
它在main.py上有一个用于创建日志的自定义中间件,这使得我的backgroundTask像正常函数一样工作,我改变了它,它工作了!

  1. @app.middleware("http")
  2. async def custom_middleware_logging(request: Request, call_next):
  3. response = await call_next(request)
  4. # logging logic...
  5. return response

相关问题