Python thread with queue duplicates items inside queue

Question

I am trying to implement a worker thread to go through a queue and add the items inside to a sql db. But I am experiencing this weird issue where even though I am definetly putting in different statements to the queue, they all become copies of each other inside the queue if I am putting them in the queue within 2 seconds.

This is the worker thread with the queue:

class DBWriterThread(threading.Thread):
    def __init__(self):
        super().__init__()
        self.q = queue.Queue()
        self.put = self.q.put

        self.start()

    def run(self):
        db_conn = None
        while True:
            statements = [self.q.get()]
            try:
                while self.q.empty() is False:
                    statements.append(self.q.get(), block=False)
            except queue.Empty:
                pass
            try:
                if statements[0] is None:
                    return
                if not db_conn:
                    db_conn = connect_to_db()
                try:
                    cursor = db_conn.cursor()
                    for statement in statements:
                        print(statement)
                        if statement is None:
                            return
                        
                        work_order = statement[0][0]
                        data = statement[0][1]

                        if work_order == 'insertTick':
                            print(f"GOT ORDER TO INSERT DATA OF SYMBOL {data['symbol']}")
                            insertDataRow(data, cursor)
                        elif work_order == 'insertTrade':
                            insertTradeSignal(data, cursor)
                        else:
                            print("Unknown work order")
                            print(work_order)
                finally:
                    db_conn.commit()
            finally:
                for _ in statements:
                    self.q.task_done()

I instantiate this thread class inside my main program (which is also a class) where I in the init method is defining it as

self.db_writer = DBWriterThread()

and then throughout the program I am doing

self.dbWriter.put((['insertTick', self.peopleTick],))

to insert data in the queue.

I believe what is happening is that I am getting multiple peopleTicks within a very short span (practically simultaneously), where after I am calling the self.dbWriter.put((['insertTick', self.peopleTick],)) with the two different peopleTicks directly after each other.

This is where I am experiencing that even though I input two different peopleTicks into the queue, when the worker thread retrieves the queue, the items inside are duplicates of the same tick. This can be stopped if I do a time.sleep(2) (not less) between I call the self.dbWriter.put but that would not work for my program and defeats the whole point of a queue. How do I solve this?

I have tried figuring out locks if that could help, but I don't know how to implement it and if that would be the solution.

Python thread with queue duplicates items inside queue

0 Answers