我有以下代码：
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
在同步而非非模式下使用 FastAPI async，我希望能够接收 POST 请求的原始、未更改的正文。
我能找到的所有示例都显示async代码，当我以正常同步方式尝试时，它们request.body()显示为协程对象。
XML当我通过向此端点发布一些内容来测试它时，我得到了一个500 "Internal Server Error".
from fastapi import FastAPI, Response, Request, Body
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.post("/input")
def input_request(request: Request):
# how can I access the RAW request body here?
body = request.body()
# do stuff with the body here
return Response(content=body, media_type="application/xml")
这对于 FastAPI 来说是不可能的吗？
注意：简化的输入请求如下所示：
POST http://127.0.0.1:1083/input
Content-Type: application/xml
<XML>
<BODY>TEST</BODY>
</XML>
我无法控制输入请求的发送方式，因为我需要替换现有的 SOAP API。
FastAPI 和 uvicorn 的新手，但我想知道为什么当我通过从命令行使用 uvicorn 启动它来运行我的“hello world”服务时，它工作正常，但是当从我的服务内部使用“uvicorn.run”方法时，服务启动，但是当我发送 GET 时，我总是得到一个404响应正文为{"detail": "Not Found"}?
这是我的代码：
import uvicorn
from fastapi import FastAPI
app = FastAPI()
uvicorn.run(app, host="127.0.0.1", port=5049)
@app.get("/")
async def root():
return {"message": "Hello World"}
总是返回 404，如下所示：
# curl http://127.0.0.1:5049/
{"detail":"Not Found"}
我的服务的输出显示：
INFO: Started server process [28612]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:5049 (Press CTRL+C to quit)
INFO: 127.0.0.1:55446 - "GET / HTTP/1.1" 404 Not Found
如果我注释掉“uvicorn.run”行，然后从命令行启动该服务（在 Windows 10 上运行）： …
各位晚上好。我正在尝试使用 FastAPI 请求将新用户添加到我的数据库中。当我尝试通过 python 控制台应用程序执行此操作时，FastAPI 向我显示以下消息：
{
'detail': [
{
'loc': ['body', 'nickname'],
'msg': 'field required',
'type': 'value_error.missing'
},
{
'loc': ['body', 'password'],
'msg': 'field required',
'type': 'value_error.missing'
},
{
'loc': ['body', 'email'],
'msg': 'field required',
'type': 'value_error.missing'
}
]
}
但是，当我执行此请求时，/docs一切正常！
这是我的 pydantic 模型：
{
'detail': [
{
'loc': ['body', 'nickname'],
'msg': 'field required',
'type': 'value_error.missing'
},
{
'loc': ['body', 'password'],
'msg': 'field required',
'type': 'value_error.missing'
},
{
'loc': ['body', 'email'],
'msg': 'field required',
'type': 'value_error.missing'
} …我正在使用 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()
我不明白FastAPI 中的可选查询参数。它与默认值为 的默认查询参数有何不同None？
如上面链接中所述，下面的示例中的可选查询参数arg1与 之间有什么区别？arg2arg2
@app.get("/info/")
async def info(arg1: int = None, arg2: int | None = None):
return {"arg1": arg1, "arg2": arg2}
我想app在我的路由器文件中获取该实例，我该怎么办？
我main.py的如下：
# ...
app = FastAPI()
app.machine_learning_model = joblib.load(some_path)
app.include_router(some_router)
# ...
现在我想在some_router的文件中使用app.machine_learning_model，我该怎么办？
我有 2 个 Pydantic 型号 (var1和var2)。该方法的输入PostExample可以接收第一模型或第二模型的数据。使用Union有助于解决此问题，但在验证过程中，它会引发第一个模型和第二个模型的错误。
如何做到在填写字段时出现错误时，仅针对某个模型返回验证器错误，而不是同时返回两个模型？（如果有帮助的话，可以通过字段A的长度来区分模型）。
主要.py
@app.post("/PostExample")
def postExample(request: Union[schemas.var1, schemas.var2]):
result = post_registration_request.requsest_response()
return result
模式.py
class var1(BaseModel):
A: str
B: int
C: str
D: str
class var2(BaseModel):
A: str
E: int
F: str
我有一个相对简单的 FastAPI 应用程序，它接受查询并从 ChatGPT 的 API 流回响应。ChatGPT 正在流回结果，我可以看到它在输入时被打印到控制台。
不工作的是StreamingResponse通过 FastAPI 返回。相反，响应会一起发送。我真的不知道为什么这不起作用。
这是 FastAPI 应用程序代码：
import os
import time
import openai
import fastapi
from fastapi import Depends, HTTPException, status, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse
auth_scheme = HTTPBearer()
app = fastapi.FastAPI()
openai.api_key = os.environ["OPENAI_API_KEY"]
def ask_statesman(query: str):
#prompt = router(query)
completion_reason = None
response = ""
while not completion_reason or completion_reason == "length":
openai_stream = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": query}],
temperature=0.0,
stream=True,
)
for …我希望我的 FastAPI 应用程序在出现以下情况时重新加载.csv我希望我的 FastAPI 应用程序在同一目录中的文件发生更改
我尝试了以下命令，但它不起作用。
uvicorn main:app --reload --reload-include *.csv
有谁知道这个问题的解决方案？
fastapi ×10
python ×8
uvicorn ×3
pydantic ×2
starlette ×2
asynchronous ×1
openai-api ×1
python-3.x ×1
streaming ×1