我正在使用 FastAPI 在 python 中创建一个服务器，并且我想要一个与我的 API 无关的函数，每 5 分钟在后台运行一次（例如从 API 检查内容并根据响应打印内容）
我尝试创建一个运行该函数的线程start_worker，但它不打印任何内容。
有谁知道该怎么做？
def start_worker():
print('[main]: starting worker...')
my_worker = worker.Worker()
my_worker.working_loop() # this function prints "hello" every 5 seconds
if __name__ == '__main__':
print('[main]: starting...')
uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)
_worker_thread = Thread(target=start_worker, daemon=False)
_worker_thread.start()
我想app在我的路由器文件中获取该实例，我该怎么办？
我main.py的如下：
# ...
app = FastAPI()
app.machine_learning_model = joblib.load(some_path)
app.include_router(some_router)
# ...
现在我想在some_router的文件中使用app.machine_learning_model，我该怎么办？
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
在 FastAPI 框架内：
虽然请求数据当然可以作为参数传递，但我想知道函数是否可以在不传递参数的情况下访问有关当前请求的信息。
免责声明：我不认为全局访问请求数据是一个好的做法，但我有一个用例，能够做到这一点会非常好。
我想定义一个从文本文件生成的 dict 变量一次，并用它来响应 API 请求。
该变量应该始终可用，直到服务器运行结束。
在下面的示例中：
from fastapi import FastAPI
import uvicorn
app = FastAPI()
def init_data(path):
print("init call")
data = {}
data[1] = "123"
data[2] = "abc"
return data
data = init_data('path')
@app.get('/')
def example_method():
# data is defined
return {'Data': data[1]}
if __name__ == '__main__':
uvicorn.run(f'example_trouble:app', host='localhost', port=8000)
我会得到：
init call
init call
INFO: Started server process [9356]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)
向 …
我有一个使用Uvicornwith的应用程序FastAPI。我还打开了一些连接（例如到MongoDB）。SIGINT一旦出现某些信号（ 、SIGTERM和） ，我想优雅地关闭这些连接SIGKILL。
我的server.py文件：
import uvicorn
import fastapi
import signal
import asyncio
from source.gql import gql
app = fastapi.FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
app.mount("/graphql", gql)
# handle signals
HANDLED_SIGNALS = (
signal.SIGINT,
signal.SIGTERM
)
loop = asyncio.get_event_loop()
for sig in HANDLED_SIGNALS:
loop.add_signal_handler(sig, _some_callback_func)
if __name__ == "__main__":
uvicorn.run(app, port=6900)
不幸的是，我尝试实现这一目标的方法不起作用。当我尝试Ctrl+C在终端中时，没有任何反应。我相信这是因为Uvicorn是在不同的线程中启动的......
这样做的正确方法是什么？我注意到了uvicorn.Server.install_signal_handlers()这个功能，但使用起来并不幸运......
我有一个 API 端点（FastAPI / Uvicorn）。除此之外，它还向另一个 API 请求信息。当我使用多个并发请求加载 API 时，我开始收到以下错误：
h11._util.LocalProtocolError: can't handle event type ConnectionClosed when role=SERVER and state=SEND_RESPONSE
在正常环境中，我会利用request.session，但我知道它不是完全线程安全的。
因此，在 FastAPI 等框架内使用请求的正确方法是什么，其中多个线程将requests同时使用该库？
我有一个 api.py 文件，其中包含以下内容：
from fastapi import FastAPI
import logging
import uvicorn
app = FastAPI(title="api")
LOG = logging.getLogger(__name__)
LOG.info("API is starting up")
LOG.info(uvicorn.Config.asgi_version)
@app.get("/")
async def get_index():
LOG.info("GET /"
return {"Hello": "Api"}
该应用程序在本地运行：
uvicorn api:app --reload
INFO: Will watch for changes in these directories: ['/Users/user/code/backend/api']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [44258] using StatReload
INFO: Started server process [44260]
INFO: Waiting for application startup.
INFO: Application startup complete.
它不记录任何启动消息。
稍后向 api 发送 http 请求时： …