有没有可能混合pydantic BaseModels和Starlette请求？
有没有可能混合pydantic BaseModels和Starlette请求？
提问于 2021-11-16 17:48:12
目标
我的目标是构建一个简单的api来使用fastapi和John来破解密码散列。目前，我的api只接受一个Post请求，其中包含要破解的散列和一些关于原始密码的可选信息(最小长度、最大长度等)。最终，它会将这个散列发送到运行容器化的John the Ripper的后端集群，以破解该散列。为了包含我希望出现在Post请求中的所有信息，我创建了一个包含所需信息的BaseModel子类(参见下面的代码)。
我在哪里
我想实施速率限制，以便每个ip地址和每分钟或小时只允许一定数量的呼叫。经过一些研究，我决定使用slowapi提供的解决方案，如下所示：
from fastapi import FastAPI
from enum import Enum
from pydantic import BaseModel
from typing import Optional
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from starlette.requests import Request
class HashType(str, Enum):
md5 = "md5"
sha1 = "sha1"
class HashRequest(BaseModel, Request):
hash: str
type: Optional[HashType] = None
min_length: Optional[int] = None
max_length: Optional[int] = None
special_chars: Optional[bool] = None
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.post("/")
@limiter.limit("10/minute")
@limiter.limit("100/hour")
async def send_hash(request: HashRequest):
## TODO: communicate with backend
return {"message":"request recieved", "hash":request.hash}slowapi需要将请求参数显式地传递给我的端点，并且它的类型是starlette.requests。我的解决方案是使用多重继承，让HashRequest继承BaseModel和Request。
当我尝试向api发送一个Post请求时，我得到了错误: AttributeError：' request‘对象没有'hash’属性。
发送请求的命令：
curl -X 'POST' 'http://127.0.0.1:8000/' -H 'accept: application/json' -H 'Content-Type: application/json' -d '{
"hash":"foo"
}'回答 1
Stack Overflow用户
发布于 2021-11-16 18:20:19
Here您可以查看请求对象主体中的内容
有几个不同的接口可用于返回请求的正文：
以字节为单位的请求正文: await request.body()
请求正文，解析为表单数据或多部分: await request.form()
请求主体，解析为JSON: await request.json()
您还可以使用async for语法以流的形式访问请求正文：
在您的情况下，您应该使用：
my_hash = request.json['hash']如果您想要使用pydantic模型：
class HashRequest(BaseModel):
hash: str
type: Optional[HashType] = None
min_length: Optional[int] = None
max_length: Optional[int] = None
special_chars: Optional[bool] = None@app.get("/")
@limiter.limit("10/minute")
@limiter.limit("100/hour")
def get_role(hash_obj: HashRequest):
## TODO: communicate with backend
return {"message": "request recieved", "hash": hash_obj.hash}页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/69993837
复制相关文章
相似问题
腾讯云开发者
