是否可以混合使用 pydantic BaseModels 和 Starlette Request？

Question

目标

我的目标是构建一个简单的 api 来使用 fastapi 和 John the ripper 破解密码哈希。 目前，我的 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 的 Request。我的解决方案是使用多重继承，使HashRequest继承自BaseModel和Request。

当我尝试向 api 发送 Post 请求时，出现错误：AttributeError: 'Request' object has no attribute 'hash'。

发送请求的命令：

curl -X 'POST'   'http://127.0.0.1:8000/'   -H 'accept: application/json'   -H 'Content-Type: application/json'   -d '{
    "hash":"foo"
}'

Answer 1

在这里你可以看到你的请求对象Body中有什么内容

有几种不同的接口用于返回请求正文：

请求正文以字节为单位：

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}