Python3Tornado限制IP在时间段内请求次数

原因&＃xff1a;有个接口服务需要限制接口请求次数。

处理逻辑&＃xff1a;

1、新建空列表&＃xff1b;

2、有IP 访问则记录 IP、时间、次数&＃xff1b;  eg&＃xff1a; [{&＃039;192.168.1.124&＃039;: {&＃039;2020-12-08 17:00:33&＃039;: 10}}, {&＃039;192.168.1.125&＃039;: {&＃039;2020-12-08 17:00:33&＃039;: 9}}]

3、判断当前时间和列表中的IP访问时间是否相同&＃xff1a;

      ①、相同则增加访问次数&＃xff1b;

      ②、不同则更改原有的访问时间和次数&＃xff1b;

4、同一秒访问10次则限制3分钟后再访问&＃xff1b;

5、防止列表一直存储着信息&＃xff0c;需要在限制的时候清空&＃xff0c;防止占满内存。

代码&＃xff1a;

import datetime # 时间包
#第三方 格式化时间包 pip install python-dateutil
from dateutil.relativedelta import relativedelta
expire_time &＃061; &＃039;&＃039; # 过期时间
iplist &＃061; [] # 存放访问的IP 格式 [{&＃039;192.168.1.124&＃039;: {&＃039;2020-12-08 17:00:33&＃039;: 10}}, {&＃039;192.168.1.125&＃039;: {&＃039;2020-12-08 17:00:33&＃039;: 9}}]
def counts(host):
global expire_time, iplist
key_list &＃061; [] # 临时存储列表中的IP键
index_list &＃061; {} # 临时存储列表中的下标值
datetimes &＃061; datetime.datetime.now()
nowtime &＃061; datetimes.strftime("%Y-%m-%d %H:%M:%S") # 转化成字符串便于存储
print(&＃039;nowtime&＃039;, nowtime)
if expire_time &＃061;&＃061; &＃039;&＃039;: # 判断无过期时间
if iplist: # 判断有访问IP
for index, item in enumerate(iplist): # 将原有的访问IP遍历出来
tmp_key &＃061; tuple(item.keys())[0] # 取出第一层字典键
if tmp_key not in key_list: # 不在临时IP存储列表中
key_list.append(tmp_key) # 不在临时IP列表则加入
tmp_key_dict &＃061; {tmp_key: index} # 拼接新临时存储IP下标&＃xff0c;方便从IPlist中取值
index_list.update(tmp_key_dict) # 更新到临时存储IP字典中
print(&＃039;iplist&＃039;, iplist)
print("key_list", key_list)
print("index_list", index_list)
if host in key_list: # 在列表中则判断时间和次数
index &＃061; index_list[host] # 取IP在IPList中的下标
item &＃061; iplist[index] # 取出IPList对应的字典值
times_key &＃061; tuple(item[host].keys())[0] # 取出IPList的第二层时间键
# 转为时间格式进行时间间隔计算
time_key &＃061; datetime.datetime.strptime(times_key, &＃039;%Y-%m-%d %H:%M:%S&＃039;)
# print("时间格式",type(datetimes - time_key),datetimes - time_key)
# >>> 时间格式 0:00:00.213644
# datetime.timedelta 格式数据通过 .seconds取相差的秒数&＃xff0c;取分钟及其他类型会报错&＃xff0c;有了解的大佬欢迎在评论区指教&＃xff01;
if (datetimes - time_key).seconds &＃061;&＃061; 0: # 时间相同 则判断是否超过10次
if item[host][times_key] >&＃061; 10: # 同一秒访问等于或大于10 就不允许请求接口了
# 设置过期时间
expire_time &＃061; (datetimes &＃043; relativedelta(minutes&＃061;&＃043;3)).strftime("%Y-%m-%d %H:%M:%S")
iplist &＃061; [] # 重置记录列表
# 返回限制结果
return {&＃039;code&＃039;: -1, &＃039;msg&＃039;: "Error", &＃039;data&＃039;: "访问频率过快&＃xff0c;请3分钟后再访问!"}
else: # 不等于10次 则增加该IP请求次数
num &＃061; item[host].get(times_key, 1) &＃043; 1 # 取出该字典的次数&＃043;1
item[host][times_key] &＃061; num # 重新赋值给该IP
else: # 时间不同则重新给IP计数
item[host] &＃061; {nowtime: 1} # 重置该IP访问次数
else: # 表示第一次访问接口
tmp_data &＃061; {host: {nowtime: 1}} # 构建该IP的访问次数
iplist.append(tmp_data) # 添加到列表中
else: # 判断无访问IP即表示第一次访问接口
tmp_data &＃061; {host: {nowtime: 1}} # 构建该IP的访问次数
iplist.append(tmp_data) # 添加到列表中
else: # 限制时间已过就可以访问接口
if nowtime &＃061;&＃061; expire_time:
tmp_data &＃061; {host: {nowtime: 1}} # 构建该IP的访问次数
iplist.append(tmp_data) # 添加到列表中
else: #限制时间未过则需要返回限制结果&＃xff0c;否则返回None
iplist &＃061; [] # 重置记录列表
return {&＃039;code&＃039;: -1, &＃039;msg&＃039;: "Error", &＃039;data&＃039;: "访问频率过快&＃xff0c;请3分钟后再访问!"}
if __name__ &＃061;&＃061; &＃039;__main__&＃039;:
for i in range(11):
print(&＃039;遍历的次数&＃xff1a;【%s】&＃039; % i)
print(counts("192.168.1.124"))
print(counts("192.168.1.125"))

本文地址：https://blog.csdn.net/qq_42142258/article/details/110877394