Check if we find new value from a request

Question

I have created a script where I add a timestamp to each value that has been found with a code that I have written:

import random
from datetime import datetime, timedelta
from typing import Dict, List
import time

class RequestFilter:
    """Tracks requests and filters them to prevent hammering."""

    def __init__(self, cooldown: timedelta):
        self._cooldown = cooldown
        self._requests: Dict[str, datetime] = {}

    def filter(self, requests: List[str], time: datetime) -> List[str]:
        """Filter requests to only those that haven't been made
        previously within our defined cooldown period."""
        # Get filtered set of requests.
        filtered = [
            r for r in list(set(requests))
            if (
                    r not in self._requests or time - self._requests[r] >= self._cooldown
            )
        ]
        # Refresh timestamps for requests we're actually making.
        for r in filtered:
            self._requests[r] = time
            
        print(self._requests)
        return filtered


if __name__ == '__main__':
    from time import sleep
    request_filter = RequestFilter(timedelta(minutes=5))

    firstReq = []
    for _ in range(random.randint(1,5)):
        firstReq.append(f"US {random.randint(1, 10)}")

    for _ in range(100):

        newReq = []
        for _ in range(random.randint(2, 8)):
            newReq.append(f"US {random.randint(1, 10)}")

        if len(newReq) > len(firstReq):
            print(request_filter.filter(newReq, datetime.now()), datetime.now())
            sleep(1)
            firstReq = newReq
        else:
            print("Length is not bigger, testing again in 3 sec...")
            time.sleep(3)
            firstReq = newReq

As you can see at the very bottom i'm checking if the list size from previous request is less than newest request (at this momentit just random function but it would be reading from a HTML later on) and if it is, that means that something has been added to a webpage and we want to see what value has been added. If the value has already a timestamp then we check "filter" and see if it has been over 5 minutes difference in the timestamp and if its true then we should say "New value has been found!"

However my improvements in general here is that I am not quite happy with the way that I check for length of firstReq vs newReq. and reason if that could be etc if I request a page and it has US 3 and then the newReq has value US 6 but it still will have 1 > 1 which is false but still different value which will not print due to the 1 > 1. My question is, how can I improve the code that I could skip the < function and check maybe the sizes directly?

Answer 1

Global code

Move the code after your __main__ check into a function, since it's still in global scope.

PEP8

firstReq should be first_req, and the same for new_req.

Dict comprehension

I would rearrange your filtered comprehension to be only a dict, and to use dict.update. In other words,

filtered = {r: self._requests.get(r) for r in requests}

self._requests.update({
    r: time
    for r, orig_time in filtered.items()
    if orig_time is None or time - orig_time >= self._cooldown
})

List comprehension

    firstReq = []
    for _ in range(random.randint(1, 5)):
        firstReq.append(f"US {random.randint(1, 10)}")

can be

first_req = [
    f"US {random.randint(1, 10)}"
    for _ in range(random.randint(1, 5))
]

Diagnostic printing

    print(self._requests)

is out-of-place. Remove it, and if you still want this to happen, issue an equivalent statement from main(), perhaps printing the entire RequestFilter and overriding __str__ to return pprint.pformat(self._requests).

Change detection

Apparently what you actually want is to return a boolean that is true if self._requests has changed. This can be done like:

filtered = {r: self._requests.get(r) for r in requests}

old_requests = dict(self._requests)

self._requests.update({
    r: time
    for r, orig_time in filtered.items()
    if orig_time is None or time - orig_time >= self._cooldown
})

return old_requests == self._requests