HTTP如何在 FastAPI 的请求之间共享变量值?例如,我有一个POST请求,其中我获取一些音频文件,然后将它们的信息转换为 Pandas Dataframe。我想在请求Dataframe中发送它GET,但我无法访问Dataframe请求GET范围。
@app.post(
path="/upload-audios/",
status_code=status.HTTP_200_OK
)
async def upload_audios(audios: list[UploadFile] = File(...)):
filenames = [audio.filename for audio in audios]
audio_data = [audio.file for audio in audios]
new_data = []
final_data = []
header = ["name", "file"]
for i in range(len(audios)):
new_data = [filenames[i], audio_data[i]]
final_data.append(new_data)
new_df = pd.DataFrame(final_data, columns=header)
return f"You have uploaded {len(audios)} audios which names are: {filenames}"
@app.get("/get-dataframe/")
async def get_dataframe():
pass
Run Code Online (Sandbox Code Playgroud) 我正在上一堂发送通知的课。初始化时,它涉及到与通知服务器的连接,这非常耗时。我在 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
Run Code Online (Sandbox Code Playgroud)
这里,通知客户端是全局声明的。我可以在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
Run Code Online (Sandbox Code Playgroud) 如何从依赖项内部修改请求?基本上我想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."}
Run Code Online (Sandbox Code Playgroud) 我有一个用于测试/开发目的的 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()
Run Code Online (Sandbox Code Playgroud)