反爬虫策略手把手教你使用FastAPI来限制接口的访问速率

Python进阶者

发布于 2021-01-05 15:27:54

3.4K00

代码可运行

文章被收录于专栏：Python爬虫与数据挖掘Python爬虫与数据挖掘

运行总次数：0

代码可运行

在网络爬虫的过程中，我们都会遇到各种各样的反爬虫，封禁IP和账号，设置验证码，前端加密，浏览器指纹，甚至输出假数据来等等都是可能出现的反爬手段，这些我们今天一个也不会谈及，而是谈谈一种叫访问速率限制的手段。

对于服务端而言，有时候会碰到这么一个场景：某个接口需要在某个时间段内设置最高的访问次数来降低服务器的压力，比如之前用的某度的一些接口，一分钟内访问次数过高就会返回失败，等上个2分钟就又可以返回了。目的就是为了防止开发人员或者爬虫，甚至是恶意请求对服务器无限制的访问，降低服务器开支，因为一般的用户的请求是不会这么频繁的

Ratelimiter

python 中使用 Ratelimiter 来限制某方法的调用次数，用法如下

import time
from ratelimiter import RateLimiter

def limited(until):
    duration = int(round(until - time.time()))
    print('Rate limited, sleeping for {:d} seconds'.format(duration))
# 3秒之内只能访问2次
rate_limiter = RateLimiter(max_calls=2, period=3, callback=limited)

for i in range(3):
    with rate_limiter:
        print('Iteration', i)

输出结果如下

Iteration 0
Iteration 1
Rate limited, sleeping for 3 seconds
Iteration 2

看到程序如期打印， callback 指定了超出指定次数是回调方法

达到了预期的要求

asyncio 异步中的使用

import asyncio
import time

from ratelimiter import RateLimiter

async def limited(until):
    duration = int(round(until - time.time()))
    print('Rate limited, sleeping for {:d} seconds'.format(duration))

async def coro():
    rate_limiter = RateLimiter(max_calls=2, period=3, callback=limited)
    for i in range(3):
        async with rate_limiter:
            print('Iteration', i)

loop = asyncio.get_event_loop()
loop.run_until_complete(coro())

执行结果是一致的, 在一般的 python 方法里面用 Ratelimiter 是没有问题的

Slowapi

对于网络请求的访问速率限制,我建议使用 Slowapi 库，Slowapi相对灵活易用，不必考虑更多的因素。在 fastapi 和 flask 中使用也是得心应手，当然flask框架也有第三方扩展，这个自不必说，django也有自带的限制访问速率的库，而 fastapi 相对比较新，扩展库相对匮乏，在一个偶然的机会看到 Slowapi 的源码, 这是一个非常不错的选择，如果有机会，我会把它封装成 fastapi 框架的另一个插件，继成更多的功能，名字可以是 fastapi-slowapi 之类的

来看看具体的用法

# -*- coding: utf-8 -*
# @Time : 2020/11/11 11:09
from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

# 初始化 slowapi，注册进 fastapi
limiter = Limiter(key_func=get_remote_address)
FastAPI().state.limiter = limiter
FastAPI().add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

具体调用方法

# -*- coding: utf-8 -*
# @Time : 2020/11/11 11:09

# 一小时内只能调用该接口 4 次
@limiter.limit("4/hour")
async def startSpider(*, request: Request, d_obj: dict):
    if request.method == "POST":
        statusDict = get_spider_status(shopId=d_obj.get("shopId"))
        if not statusDict.get("data"):
            return resp_422(message='程序正在抓取中，请勿重复调度')
        try:
            result = all_run(d_obj.get("shopId"))
            return result
        except:
            return resp_401()
    return "这是一个GET请求"