8
\$\begingroup\$

I am making a Reddit bot and to prevent my bot from posting to the same comment twice, instead of making a database, I decided I would rather create a temporary list (or "queue") that once appended to, checks to ensure the maximum length of the queue hasn't been exceeded. Once exceeded, however, the queue is cleared.

class TemporaryQueue:
    def __init__(self, max_items=10):
        self._queue = []
        self.max_items = max_items

    def __repr__(self):
        return "<TemporaryQueue max_items={}>".format(self.max_items)

    def __iter__(self):
        yield from self._queue

    def __eq__(self, other):
        return other in self._queue

    def _exceed_check(self):
        if len(self._queue) > self.max_items:
            return True
        return False

    def append(self, value):
        if self._exceed_check():
            self._queue.clear()

        return self._queue.append(value)

Example usage:

for submission in reddit.subreddit("funny").stream.submissions():
    queue = TemporaryQueue()

    if submission.id not in queue:
        if "foo" in title:
            submission.reply("Bar!")
            queue.append(submission.id)
New contributor
docyoda is a new contributor to this site. Take care in asking for clarification, commenting, and answering. Check out our Code of Conduct.
\$\endgroup\$
  • \$\begingroup\$ You are using a database, just a very limited in-memory database. Also, you might want to retain the last n recently used items instead of discarding all the items when the limit is reached. \$\endgroup\$ – David Conrad Dec 3 at 18:05
  • \$\begingroup\$ It will post to the same comment twice, if the queue happens to clear itself in between. \$\endgroup\$ – user253751 Dec 3 at 18:24
13
\$\begingroup\$

collections.deque

Why not use a collections.deque with a max_len?

from collections import deque

for submission in reddit.subreddit("funny").stream.submissions():
    queue = deque(max_len=10)

    if submission.id not in queue:
        if "foo" in title:
            submission.reply("Bar!")
            queue.append(submission.id)
\$\endgroup\$
6
\$\begingroup\$

I made a few adjustments (simplified the return handling for max queue size) and added in a dequeue method:

class TemporaryQueue:
    def __init__(self, max_items=10):
        self._queue = []
        self.max_items = max_items

    def __repr__(self):
        return "<TemporaryQueue max_items={}>".format(self.max_items)

    def __iter__(self):
        yield from self._queue

    def __eq__(self, other):
        return other in self._queue

    def _full(self):
        return len(self._queue) >= self.max_items

    def enqueue(self, value):
        if self._full():
            self._queue.clear()

        return self._queue.append(value)

    def dequeue(self):
        return self._queue.pop()

A queue doesn't seem like a good data structure for this problem. A ring buffer might be more applicable; it creates a fixed size list of elements (say 10) and when there isn't enough room left, it overwrites the oldest one. Depending on the implementation, the entire buffer can be searched to see if it contains a string (taken directly from https://www.oreilly.com/library/view/python-cookbook/0596001673/ch05s19.html):

class RingBuffer:
    """ class that implements a not-yet-full buffer """
    def __init__(self,size_max):
        self.max = size_max
        self.data = []

    class __Full:
        """ class that implements a full buffer """
        def append(self, x):
            """ Append an element overwriting the oldest one. """
            self.data[self.cur] = x
            self.cur = (self.cur+1) % self.max
        def get(self):
            """ return list of elements in correct order """
            return self.data[self.cur:]+self.data[:self.cur]

    def append(self,x):
        """append an element at the end of the buffer"""
        self.data.append(x)
        if len(self.data) == self.max:
            self.cur = 0
            # Permanently change self's class from non-full to full
            self._ _class_ _ = self._ _Full

    def get(self):
        """ Return a list of elements from the oldest to the newest. """
        return self.data

# sample usage
if __name__=='__main__':
    x=RingBuffer(5)
    x.append(1); x.append(2); x.append(3); x.append(4)
    print x.__class__, x.get(  )
    x.append(5)
    print x.__class__, x.get(  )
    x.append(6)
    print x.data, x.get(  )
    x.append(7); x.append(8); x.append(9); x.append(10)
    print x.data, x.get(  )
\$\endgroup\$
6
\$\begingroup\$

This is a weird unique queue. It does not actually guarantee uniqueness on its own:

queue = TemporaryQueue()
for x in range(10):
    queue.append(x)
queue.append(0)
print(list(queue))
# [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0]

And it does not guarantee that all of the elements are even there:

queue = TemporaryQueue()
for x in range(10):
    queue.append(x)
print(list(queue))
# [11]

If I add more than max_items elements to a queue, I expect the queue to always grow until it reaches its maximum size and then stay constant (in size). Not shrink again to one element.

The max_items argument is also not followed correctly:

queue = TemporaryQueue()
for x in range(11):
    queue.append(x)
print(len(list(queue)))
# 11

Your __eq__ method is not needed. It looks like you wanted to implement __contains__ in order to be able to do x in queue. Your code still works because Python just uses x in list(self), which in this case is just as inefficient as your implementation.

In your append method you do return self._queue.append(value). Since list.append modifies the list inplace, it returns None, which your method will implicitly return anyways. Instead just have the call in its line as the last line of the method

On the other hand, your capacity check can be simplified by putting the call in the return:

def _exceed_check(self):
    return len(self._queue) > self.max_items

You might want to add a extend method for convenience. A simple implementation such as this would already suffice:

def extend(self, values):
    for x in values:
        self.append(x)

If you want truly unique comments, either collect them in a set beforehand, or use the itertools recipe unique_everseen if you want to process them as they come in. Of course, this needs \$\mathcal{O}(n)\$ memory, where \$n\$ is the number of unique comments.

If you want an efficient maximum size queue from which you can get elements from both sides, use a collections.deque, as recommended in the answer by @RootTwo.

\$\endgroup\$

Your Answer

docyoda is a new contributor. Be nice, and check out our Code of Conduct.

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.