本篇博客我们实现的案例是 IP 限制反爬，翻译过来就是每个 IP 在规定时间内限制访问次数。
例如，可以限制单 IP 每秒访问 5 次，超过之后就会返回 403 错误。

Flask 实现 IP 限制

• • 使用 Flask 插件
  • 自定义中间件限制 IP
  • 自定义请求钩子

使用 Flask 插件

实战中可以使用第三方模块 Flask-Limiter（Flask 限流器） 来实现 IP 限制。在 PyCharm 中直接安装即可。

使用命令行如下所示：

pip install Flask-Limiter

然后就可以在 Flask 应用中使用 Flask-Limiter 插件，提前建立相关视图函数。

视图函数在 app/school/index.py 文件中；
前端模板文件复制 templates/school/ajax_list.html 文件为 ajax_list3.html 即可。

原接口请求地址是 /ss/api2，现在修改为 /ss/api3，同时在 index.py 中复制原接口函数。

@s.route('api3')
def school_api3():page = int(request.args.get("page", 1))pagination = pagination_object(page)return jsonify(pagination)

后续我们的核心逻辑都将在上述函数进行改写，导入 Flask-Limiter 插件。

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address,get_ipaddr

上述代码在 app/__init__.py 中进行实现，实例化操作也在该文件创建。

limiter = Limiter(app, key_func=get_ipaddr)

使用的时候，只需要引入该对象，然后添加装饰器即可。

# 从 app 中导入 limiter 对象
from app import limiter@s.route('api3')
@limiter.limit("3/second")
def school_api3():page = int(request.args.get("page", 1))pagination = pagination_object(page)return jsonify(pagination)

上述代码 @limiter.limit("3/second") 表示限制每秒3次请求，有两种方式表示速率限制，如下所示。

“100 per day”、“50 per hour”、“20 per minute”、“1 per second”
“100/day”、“50/hour”、“20/minute”、“1/second”

装饰器可以单一修饰，也可以多个修饰，分别如下所示。

• 单一修饰：限制字符串可以是单个限制，也可以是定界符分隔的字符串。

@app.route("api3")
@limiter.limit("100/day;10/hour;1/minute")
def my_route()

• 多个装饰器：限制字符串可以是单个限制，也可以是定界符分隔的字符串，也可以是两者的组合。

@app.route("api3")
@limiter.limit("100/day")
@limiter.limit("10/hour")
@limiter.limit("1/minute")
def my_route():

除此之外，在实例化限速器对象的时候，还用到了限制域内容，特指根据什么进行限制。

• get_remote_address：根据请求的 remote_address；
• get_ipaddr：根据请求 X-Forwarded-For 标头中的最后IP地址。

接下来我们需要拿到生成环境做一下测试，查看一下捕获到的 IP 是否正确，在爬虫训练冲项目中，可以正确获取到 IP 值，如果你使用了反向代理，那么 Flask-Limiter 将使用默认的 key_func，即 get_remote_address 来获取客户端的 IP 地址。 在这种情况下，Flask-Limiter 将会看到反向代理的 IP 地址，而不是实际客户端的 IP 地址。

为了解决这个问题，你可以使用自定义的 key_func 函数，用来检索客户端的真实 IP 地址。 例如：

from flask import requestdef get_real_ip():if request.headers.getlist("X-Forwarded-For"):return request.headers.getlist("X-Forwarded-For")[0]return request.remote_addrlimiter = Limiter(app, key_func=get_real_ip)

这样，Flask-Limiter 就会使用客户端的真实 IP 地址来进行限制，而不是反向代理的 IP 地址。

请注意，如果你使用了多个反向代理，那么上述 get_real_ip() 函数可能无法获取客户端的真实 IP 地址。 在这种情况下，你可能需要检查所有的 “X-Forwarded-For” 头，以找到客户端的真实 IP 地址。 例如：

def get_real_ip():x_forwarded_for = request.headers.getlist("X-Forwarded-For")if x_forwarded_for:return x_forwarded_for[-1]return request.remote_addr

这样，你就可以使用 Flask-Limiter 来限制客户端的真实 IP 地址，即使在反向代理的情况下也是如此。

实战中有时候需要记录 IP 访问此处，判断该 IP 访问次数是否达到上线，Limiter 默认使用的是内存记录，这在生产环境中显然是不现实的，所以这里补充一个 小知识点：Flask-Limiter 使用 redis 作为数据储存容易。

自定义中间件限制 IP

你也可以自己实现一个中间件来实现 IP 限制。

中间件是一个可以在请求和响应之间插入的代码，通常用于实现某些功能，比如认证、IP 限制等。

在 Flask 中，中间件可以通过装饰器的方式使用，例如：

def ip_limiter(app):def middleware(next):def wrapper(*args, **kwargs):# 在这里处理 IP 限制逻辑return next(*args, **kwargs)return wrapperreturn middlewareapp.register_middleware(ip_limiter)

后续操作是在 middleware() 函数中实现 IP 限制逻辑即可。

自定义请求钩子

Flask 还提供了一种在请求之前插入代码的机制，叫做请求钩子。使用方法如下：

@app.before_request
def ip_limiter():# 在这里处理 IP 限制逻辑pass

在 ip_limiter() 函数中，你可以获取请求的 IP 地址，然后根据你的限制规则来决定是否允许这个请求继续执行。

例如，你可以使用 Flask 提供的 request 对象来获取请求的 IP 地址：

@app.before_request
def ip_limiter():ip = request.remote_addr# 在这里处理 IP 限制逻辑pass

注意，如果你的应用部署在有反向代理的环境中，那么 request.remote_addr 可能不是真实的客户端 IP，而是反向代理的 IP。在这种情况下，你需要使用特定的方法来获取客户端的真实 IP。

最后，如果你希望限制某个 IP 访问次数，你可以使用一个字典来记录每个 IP 的访问次数，每次请求时递增计数器，然后判断是否超过限制。

ip_counter = {}@app.before_request
def ip_limiter():ip = request.remote_addrif ip not in ip_counter:ip_counter[ip] = 0ip_counter[ip] += 1if ip_counter[ip] > 10:# 超过限制，拒绝请求abort(429)

上述代码实现的结果是：每个 IP 每分钟最多访问 10 次。

本案例到此结束，已更新到 爬虫训练场 欢迎大家访问学习。
项目同步到代码仓库 https://gitcode.net/hihell/spider_playground

📢📢📢📢📢📢
💗 你正在阅读 【梦想橡皮擦】 的博客
👍 阅读完毕，可以点点小手赞一下
🌻 发现错误，直接评论区中指正吧
📆 橡皮擦的第 820 篇原创博客

从订购之日起，案例 5 年内保证更新

• ⭐️ Python 爬虫 120，点击订购 ⭐️
• ⭐️ 爬虫 100 例教程，点击订购 ⭐️