在处理 FastAPI 请求时，我有一个受 CPU 限制的任务要对列表的每个元素执行。我想在多个 CPU 内核上进行此处理。
在 FastAPI 中执行此操作的正确方法是什么？我可以使用标准multiprocessing模块吗？到目前为止，我发现的所有教程/问题仅涵盖 I/O 绑定任务，如 Web 请求。
我们已经使用 FastAPI 创建了一个服务。当我们的服务启动时，它会创建一些 Python 对象，然后端点使用这些对象来存储或检索数据。
生产中的 FastAPI 从多个工人开始。我们的问题是每个工人创建自己的对象而不是共享一个。
下面的脚本显示了我们正在做的（简化的）示例，尽管在我们的例子中 Meta() 的使用要复杂得多。
from fastapi import FastAPI, status
class Meta:
def __init__(self):
self.count = 0
app = FastAPI()
meta = Meta()
# increases the count variable in the meta object by 1
@app.get("/increment")
async def increment():
meta.count += 1
return status.HTTP_200_OK
# returns a json containing the current count from the meta object
@app.get("/report")
async def report():
return {'count':meta.count}
# resets the count in the meta object to 0
@app.get("/reset") …最近，我问了一个关于如何在部署的 API 中跟踪 for 循环进度的问题。这是链接。
对我有用的解决方案代码是，
from fastapi import FastAPI, UploadFile
from typing import List
import asyncio
import uuid
context = {'jobs': {}}
app = FastAPI()
async def do_work(job_key, files=None):
iter_over = files if files else range(100)
for file, file_number in enumerate(iter_over):
jobs = context['jobs']
job_info = jobs[job_key]
job_info['iteration'] = file_number
job_info['status'] = 'inprogress'
await asyncio.sleep(1)
jobs[job_key]['status'] = 'done'
@app.get('/')
async def get_testing():
identifier = str(uuid.uuid4())
context['jobs'][identifier] = {}
asyncio.run_coroutine_threadsafe(do_work(identifier), loop=asyncio.get_running_loop())
return {"identifier": identifier}
@app.get('/status/{identifier}')
async …