Masonite 含有一个流量限制特性，非常方便地在一个给定的时间窗口内，限制操作次数。

最常用的一个地方是，限制一些API端点的流量，但你也可以用来限制一个模型在这段时间里更新多少次，队列里有多少个任务，或者这个时间内发送多少邮件。

这个特性是基于应用的 Cache 特性。它会保存给予 key 的运行次数 ，并保存相对应的时间窗口。

概况

可以通过容器 application.make("rate") 或者 RateLimiter facade 使用流量限制特性。

限制一个操作，你需要：

• key 是该操作的唯一标识
• 授权的运行次数
• 达到运行次数后，需要等待多长时间

步骤

在使用 Masonite 的流量限制之前，我们需要在 providers 列表中注册 RateProvider 类：

from masonite.providers import RateProvider
# ..
PROVIDERS = [
    #..
    RateProvider,
]

限制 HTTP 请求

使用 ThrottleRequestsMiddleware 可以轻松地限制 HTTP 请求。

首先在项目中注册一个中间件，作为路由中间件：

# Kernel.py
from masonite.middleware import ThrottleRequestsMiddleware

class Kernel:
    route_middleware = {
        "web": [
            SessionMiddleware,
            LoadUserMiddleware,
            VerifyCsrfToken,
        ],
        "throttle": [ThrottleRequestsMiddleware],
    }

这个中间件需要一个参数，可以是限制关键字符串，也可以是限制器的名字。

使用限制关键字符串

使用像 100/day 的限制关键字符串，你可以 全局 地进行限制，不需要指定用户，视图或者 IP 地址。会完全地限制所有的 HTTP 请求。

可以使用的单位有： minute， hour 和 day 。

现在到路由上使用它：

# web.py
Route.post("/api/uploads/", "UploadController@create").middleware("throttle:100/day")
Route.post("/api/videos/", "UploadController@create").middleware("throttle:10/minute")

使用限制器

为了能够给每个用户进行限制，或者实现更加复杂的逻辑，你需要使用 限制器 。 限制器 是一些含有 allow(request) 方法的简单的类，每当有 HTTP 请求时，就会调用它进行限制。

Masonite 里面捆绑了一些限制器：

• GlobalLimiter
• UnlimitedLimiter
• GuestsOnlyLimiter

你也可以创建自己的限制器：

from masonite.rates import Limiter

class PremiumUsersLimiter(Limiter):

    def allow(self, request):
        user = request.user()
        if user:
            if user.role == "premium":
                return Limit.unlimited()
            else:
                return Limit.per_day(10).by(request.ip())
        else:
            return Limit.per_day(2).by(request.ip())

这里创建的限制器，对于没有登录的用户，一天只能请求 2 次；对于不是 permium 角色的用户，一天只能请求 10 次；对于 premium 角色的用户，不进行限制。
这里使用的是 by(key) 去定义如何辨别用户。

最后，可以在应用的 provider 中注册你的限制器：

from masonite.facades import RateLimiter
from masonite.rates import GuestsOnlyLimiter
from app.rates import PremiumUsersLimiter

class AppProvider(Provider):

    def register(self):
        RateLimiter.register("premium", PremiumUsersLimiter())
        # 注册另外一个限制器，使获得授权的用户没有访问限制
        # 对于没有登录的用户，一天 2 个请求
        RateLimiter.register("guests", GuestsOnlyLimiter("2/hour"))

现在就可以在路由中使用它：

# web.py
Route.post("/api/uploads/", "UploadController@create").middleware("throttle:premium")
Route.post("/api/videos/", "UploadController@create").middleware("throttle:guests")

现在没有授权的情况下，访问 /api 端点，将会在响应中看到下面的头部：

• X-Rate-Limit-Limit : 5
• X-Rate-Limit-Remaining : 4

达到限制的次数后，将会有另外两个头部添加到响应中， X-Rate-Limit-Reset 表示流量限制重置，让 api 端点重新允许访问的时间，以时间戳表示； Retry-After 表示剩余多少秒使流量限制进行重置：

• X-Rate-Limit-Limit : 5
• X-Rate-Limit-Remaining : 0
• X-Rate-Limit-Reset : 1646998321
• Retry-After : 500

超过限制时，会抛出一个 ThrottleRequestsException 异常，然后返回状态码为 429: Too Many Requests 和 Too Many attempts 内容的响应。

自定义响应

响应可以进行定制，提供不同的状态码，内容和头部。在限制器中，添加 get_response() 方法就能实现。

前面的例子就会修改成：

class PremiumUsersLimiter(Limiter):
    # ...

    def get_response(self, request, response, headers):
        if request.user():
            return response.view("Too many attempts. Upgrade to premium account to remove limitations.", 400)
        else:
            return response.view("Too many attempts. Please try again tomorrow or create an account.", 400)

定义限制

为了使用流量限制的特性，需要定义一些限制。一个限制就是在给定的时间内，允许访问的数量。可以是一天 100 次，或者一小时 5 次。Masonite 定义的限制来自抽象类 Limit 。

下面是两个不同的方式创建限制：

from masonite.rates import Limit

Limit.from_str("100/day")
Limit.per_day(100)  # 与上一句的意思相同

Limit.per_minute(10)
Limit.per_hour(5)
Limit.unlimited()  # 定义一个不限流的限制

为这个限制关联上 key ：

username = f"send_mail-{user.id}"
Limit.per_hour(5).by(username)

限制一个操作

这个特性允许开发者方便地限制 Python 函数的调用。下面的代码限制键为 sam 的函数，让它每个小时只能调用 3 次。

开始进行尝试：

def send_welcome_mail():
    # ...

RateLimiter.attempt(f"send_mail-{user.id}", send_welcome_mail, max_attempts=3, delay=60*60)

或者可以手动添加次数：

RateLimiter.hit(f"send_mail-{user.id}", delay=60*60)

现在可以取得访问的次数：

RateLimiter.attempts(f"send_mail-{user.id}") #== 1

取得剩余的访问次数：

RateLimiter.remaining(f"send_mail-{user.id}", 3) #== 2

现在检查是否进行了过多的访问：

if RateLimiter.too_many_attempts(f"send_mail-{user.id}", 3):
    print("limited")
else:
    print("ok")

可以重置访问的次数：

RateLimiter.reset_attempts(f"send_mail-{user.id}")
RateLimiter.attempts(f"send_mail-{user.id}") #== 0

可以取得多少秒后能够再次访问：

RateLimiter.available_in(f"send_mail-{user.id}") #== 356

取得以 UNIX 时间戳表示的能够重新进行访问的时间：

RateLimiter.available_at(f"send_mail-{user.id}") #== 1646998321

这里是个完整的用例，它会决定邮件会不会发送给用户：

class WelcomeController(Controller):

    def welcome(self, request: Request):
        user = request.user()
        rate_key = f"send_mail_{user.id}"
        if (RateLimiter.remaining(rate_key, 2)):
            WelcomeMail(user).send()
            RateLimiter.hit(rate_key, delay=3600)
        else:
            seconds = RateLimiter.available_in(rate_key)
            return "You may try again in {seconds} seconds."

本译文仅用于学习和交流目的，转载请务必注明文章译者、出处、和本文链接
我们的翻译工作遵照 CC 协议，如果我们的工作有侵犯到您的权益，请及时联系我们。

原文地址：https://learnku.com/docs/masonite/4.0/fe...

译文地址：https://learnku.com/docs/masonite/4.0/fe...