我有以下代码：
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.get("/ping")
async def ping(request: Request):
print("Hello")
time.sleep(5)
print("bye")
return {"ping": "pong!"}
如果我在本地主机上运行我的代码 - 例如http://localhost:8501/ping- 在同一浏览器窗口的不同选项卡中，我得到：
Hello
bye
Hello
bye
代替：
Hello
Hello
bye
bye
我已经阅读过有关使用的内容httpx，但仍然无法实现真正的并行化。有什么问题？
python asynchronous concurrent-processing python-asyncio fastapi
我们正在使用 Python FastAPI 编写一个 Web 服务，该服务将托管在 Kubernetes 中。出于审计目的，我们需要保存特定路由的request/的原始 JSON 正文。JSON的主体大小约为1MB，最好这不应该影响响应时间。我们怎样才能做到这一点？responserequestresponse
我想app在我的路由器文件中获取该实例，我该怎么办？
我main.py的如下：
# ...
app = FastAPI()
app.machine_learning_model = joblib.load(some_path)
app.include_router(some_router)
# ...
现在我想在some_router的文件中使用app.machine_learning_model，我该怎么办？
我们已经使用 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") …我正在上一堂发送通知的课。初始化时，它涉及到与通知服务器的连接，这非常耗时。我在 FastAPI 中使用后台任务来发送通知，因为我不想因通知而延迟响应。下面是示例代码。
file1.py:
noticlient = NotificationClient()
@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
result = add_some_tasks(data, background_tasks, noticlient)
return result
file2.py:
def add_some_tasks(data, background_tasks: BackgroundTasks, noticlient):
background_tasks.add_task(noticlient.send, param1, param2)
result = some_operation
return result
这里，通知客户端是全局声明的。我可以在file2.pyunder中初始化它add_some_tasks，但每次请求到达时它都会被初始化，这需要一些时间。有什么办法可以使用中间件在每次请求到达时重新使用它，这样就不需要每次都进行初始化。
或方法二：在类 def 中初始化通知
file1.py:
class childFastApi(FastAPI):
noticlient = NotificationClient()
app = childFastApi()
@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
result = add_some_tasks(data, background_tasks, app.noticlient)
return result
如何从依赖项内部修改请求？基本上我想test_value向请求添加一些信息（），然后能够从视图函数（在我的例子中是root()函数）获取它。
下面是一个简单的例子：
from fastapi import FastAPI, Depends, Request
app = FastAPI()
def test(request: Request):
request['test_value'] = 'test value'
@app.get("/", dependencies=[Depends(test)])
async def root(request: Request):
print(request.test_value)
return {"test": "test root path."}
我有一个用于测试/开发目的的 FastAPI 应用程序。我想要的是，到达我的应用程序的任何请求都会按原样自动发送到另一台服务器上的另一个应用程序，并具有完全相同的参数和相同的端点。这不是重定向，因为我仍然希望应用程序像往常一样处理请求并返回值。我只想向不同服务器上的不同版本的应用程序发起类似的请求，而不需要等待其他服务器的答复，以便其他应用程序获取该请求，就像原始请求发送给它一样。
我怎样才能做到这一点？以下是我用于处理请求的示例代码：
@app.post("/my_endpoint/some_parameters")
def process_request(
params: MyParamsClass,
pwd: str = Depends(authenticate),
):
# send the same request to http://my_other_url/my_endpoint/
return_value = process_the_request(params)
return return_value.as_json()
我想将文件上传到 FastAPI 后端并将其转换为 Pandas DataFrame。但是，我似乎不明白如何使用 FastAPI 的UploadFile对象来做到这一点。更具体地说，我应该将什么传递给该pd.read_csv()函数？
这是我的 FastAPI 端点：
@app.post("/upload")
async def upload_file(file: UploadFile):
df = pd.read_csv("")
print(df)
return {"filename": file.filename}