Python. Отслеживание частоты запросов к серверу по ip

Question

Задача создать сервис, который будет отслеживать количество запросов к серверу по ip-адресу. В случае превышения определенного количества запросов за определенное время, сервис должен выдавать код 429. В моем коде для выполнения данной задачи применяются глобальные переменные: TIME - время с момента первого запроса, в течении которого проводится счет количества запросов, MAX_REQ - максимально допустимое число запросов за указанное время и список IP_LIST, в котором содержится словарь, который в свою очередь содержит ip, количество запросов (count) и время первого запроса. Реализован с использованием Django. Данный код добавляет в список словарь, считает количество запросов, выдает код 429, в случае превышения допустимого числа, по истечению указанного времени удаляет словарь с ip из списка, и затем должен добавлять очередной ip. Но на этом моменте код не работает так как планировалось, после удаления словаря из списка новый словарь не добавляется.

views.py

from django.shortcuts import render
import time

IP_LIST = []
TIME = 20
MAX_REQ = 25

def count_ip(request, ip):
    _ip = next((el for el in IP_LIST if el['ip'] == ip), False)
    if _ip:
        time_now = time.time()
        if time_now - _ip['time'] > TIME:
            new_list = [el for el in IP_LIST if not el == _ip]
            return new_list
        else:
            _ip['count'] += 1
            if _ip['count'] >= MAX_REQ:
                return render(request, 'index.html', status=429)
            return IP_LIST
    else:
        now = time.time()
        IP_LIST.append({'ip': ip, 'count': 1, 'time': now})
    return IP_LIST

def get_client_ip(req):
    """
    This is used to get the user's IP from the request object.

    """
    x_forwarded_for = req.META.get('HTTP_X_FORWARDED_FOR')
    if x_forwarded_for:
        ip = x_forwarded_for.split(',')[0]
    else:
        ip = req.META.get('REMOTE_ADDR', "unknown")
    return ip


def index(request):
    ip = get_client_ip(request)
    IP_LIST = count_ip(request, ip)
    return render(request, 'index.html', context={
        'ip': IP_LIST,
    })

index.html

<h1>IP: {{ ip }}</h1>

Answer 1

не надо тебе стейта в рантайме. это будет работать только пока ты используешь сервер разработки manage.py run. как только ты перейдешь на продакшен окружение, то это перестанет работать сразу же. храни подобного рода стейт в редисе.

тем более что доступ к редису очень быстр, в редисе уже реализован механизм удаления по таймауту.

_ip = next((el for el in IP_LIST if el['ip'] == ip), False)

очень долгая штука. храни не список словарей, а словарь словарей, чтобы по ip сразу брать данные.

С редисом выглядеть будет примерно так

from redis import Redis


def check_raite_limit(remote_addr):
    time = time.time()
    redis = Redis.from_url("redis://localhost:6379")
    data = json.loads(redis.get(remote_addr) or b'{"count": 0, "time":  null}')
    # any logic
    redis.set(remote_addr, json.dumps(data), ex=5)  # 5 seconds
    return True  # или False, смотря как ты там посчитаешь

БЛОГ НА HUSL

Python. Отслеживание частоты запросов к серверу по ip

Ответы (1 шт):