我有以下 Pydantic 模型：
class Report(BaseModel):
id: int
name: str
grade: float = None
proportion: float = None
@validator('*', pre=True)
def blank_strings(cls, v):
print(v)
if v == "":
return None
return v
我的目标是能够将空字符串视为空值，但它似乎不起作用。
Report(id=5,name="Steve",grade=0.5)
创建一个实例，其中proportion=Nonebut...
Report(id=5,name="Steve",grade=0.5,proportion="")抛出 error value is not a valid float (type=type_error.float)。我怎样才能得到与第一种情况相同的结果？
如果出现 HTTPException，如何使用 FastAPI 重定向请求？
在 Flask 中我们可以这样实现：
@app.errorhandler(404)
def handle_404(e):
if request.path.startswith('/api'):
return render_template('my_api_404.html'), 404
else:
return redirect(url_for('index'))
或者在 Django 中我们可以使用 django.shortcuts：
from django.shortcuts import redirect
def view_404(request, exception=None):
return redirect('/')
我们如何使用 FastAPI 来实现这一目标？
我目前有名为 jwt 的 JWT 依赖项，它确保它在到达端点之前通过 JWT 身份验证阶段，如下所示：
sample_endpoint.py:
from fastapi import APIRouter, Depends, Request
from JWTBearer import JWTBearer
from jwt import jwks
router = APIRouter()
jwt = JWTBearer(jwks)
@router.get("/test_jwt", dependencies=[Depends(jwt)])
async def test_endpoint(request: Request):
return True
以下是使用 JWT 对用户进行身份验证的 JWT 依赖项（来源：https ://medium.com/datadriveninvestor/jwt-authentication-with-fastapi-and-aws-cognito-1333f7f2729e ）：
JWTBearer.py
from typing import Dict, Optional, List
from fastapi import HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, jwk, JWTError
from jose.utils import base64url_decode
from pydantic import BaseModel
from starlette.requests import Request …我正在尝试运行一项使用简单变压器 Roberta 模型进行分类的服务。测试时，推理脚本/函数本身按预期工作。当我将其包含在快速 API 中时，它会关闭服务器。
\nuvicorn==0.11.8\nfastapi==0.61.1\nsimpletransformers==0.51.6\ncmd : uvicorn --host 0.0.0.0 --port 5000 src.main:app\n@app.get("/article_classify")\ndef classification(text:str):\n """function to classify article using a deep learning model.\n Returns:\n [type]: [description]\n """\n\n _,_,result = inference(text)\n return result\n错误 ：
\nINFO: Started server process [8262]\nINFO: Waiting for application startup.\nINFO: Application startup complete.\nINFO: Uvicorn running on http://0.0.0.0:5000 (Press CTRL+C to quit)\nINFO: 127.0.0.1:36454 - "GET / HTTP/1.1" 200 OK\nINFO: 127.0.0.1:36454 - "GET /favicon.ico HTTP/1.1" 404 Not Found\nINFO: 127.0.0.1:36454 - "GET /docs HTTP/1.1" …python multilabel-classification fastapi uvicorn simpletransformers
我想用 FastAPI 和 Jinja 作为模板构建一个购物车，
所以我需要在会话中保存每个匿名用户的数据。
Django和flask都有内置的session功能，我们可以很容易地做到这一点。
其中一种解决方案可以使用 SQLAlchemy 会话，但 SQLAlchemy 会话不支持匿名用户，我们必须为每个会话单独创建令牌。那么我们应该使用存储的令牌来保存每个数据。
还有其他类似Django和Flask内置功能的方式吗？
我有一个用Uvicorn + FastAPI编写的 REST-API 应用程序
我想使用 PyTest 进行测试。
我想在开始测试时在夹具中启动服务器，因此当测试完成时，夹具将终止应用程序。
FastAPI 测试展示了如何测试 API 应用程序，
from fastapi import FastAPI
from starlette.testclient import TestClient
app = FastAPI()
@app.get("/")
async def read_main():
return {"msg": "Hello World"}
client = TestClient(app)
def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"msg": "Hello World"}
这不会以通常的方式使服务器联机。似乎由 client.get 命令触发的特定功能是唯一运行的东西。
我找到了这些额外的资源，但我无法让它们为我工作：
https://medium.com/@hmajid2301/pytest-with-background-thread-fixtures-f0dc34ee3c46
您将如何从 PyTest 运行 Uvicorn+FastAPI 应用程序，以便它随着测试而上升和下降？
我在 FastAPI 上有一个 API，当他请求我的页面时，我需要获取客户端的真实 IP 地址。
我很乐意使用starlette Request。但它返回我的服务器 IP，而不是客户端远程 IP。
我的代码：
@app.post('/my-endpoint')
async def my_endpoint(stats: Stats, request: Request):
ip = request.client.host
print(ip)
return {'status': 1, 'message': 'ok'}
我在做什么错？如何获得真实IP（如在Flask request.remote_addr 中）？
我是非常新的 python 和 fastapi。我正在尝试在全局级别捕获未处理的异常。所以在main.py文件中的某个地方我写在下面：
@app.exception_handler(Exception)
async def exception_callback(request: Request, exc: Exception):
logger.error(exc.detail)
但上述方法从未执行过。但是如果我编写一个自定义异常并尝试捕获它（如下所示），它运行良好。
class MyException(Exception):
#some code
@app.exception_handler(MyException)
async def exception_callback(request: Request, exc: MyException):
logger.error(exc.detail)
我已经完成了Catch 异常类型的 Exception 和 process body request #575。但是这个错误谈论访问请求正文。看到这个bug，感觉应该可以抓到Exception。FastApi 版本fastapi>=0.52.0。
提前致谢 ：）
我正在尝试使用 FastAPI 为我们的组织创建 API。它有一个用于所有身份验证的 KeyCloak 服务器，以及被认为是最佳实践的 OpenID Connect 和 JWT。
在最简单的情况下，其他人负责获取有效的 JWT 令牌，以便 FastAPI 可以简单地解码和读取用户和权限。
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
jwt_token = jwt.decode(token, key=env.keycloak_server_public_key, audience='myorg')
return jwt_token['preferred_username']
except jwt.exceptions.ExpiredSignatureError:
raise credentials_exception
生活是简单的！
但是，我确实想尝试让用户使用 Swagger 页面探索 API。我创建了这个函数，让用户使用 UI 登录：
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
login_request = requests.post(
"https://mygreatorg.com/auth/realms/master/protocol/openid-connect/token",
data={
"grant_type": "password",
"username": form_data.username,
"password": form_data.password,
"client_id": "fastapi-application",
},
)
raw_response = json.loads(login_request.content.decode('utf-8'))
raw_response['acquire_time'] …我已经部署了一个 fastapi 端点，
from fastapi import FastAPI, UploadFile
from typing import List
app = FastAPI()
@app.post('/work/test')
async def testing(files: List(UploadFile)):
for i in files:
.......
# do a lot of operations on each file
# after than I am just writing that processed data into mysql database
# cur.execute(...)
# cur.commit()
.......
# just returning "OK" to confirm data is written into mysql
return {"response" : "OK"}
我可以从 API 端点请求输出，它对我来说工作得很好。
现在，对我来说最大的挑战是知道每次迭代需要多少时间。因为在 UI 部分（那些访问我的 API 端点的人）我想帮助他们为正在处理的每个迭代/文件显示一个进度条（TIME TAKEN）。
我有什么可能的方法来实现它吗？如果是这样，请帮助我了解如何进一步处理？
谢谢你。