Assume modern Python when using Queue class

The timeout parameter actually exists since Python 2.3...

Author: Cito

tests/test_persistent_db.py

@@ -76,10 +76,7 @@ def run_queries(i):
         db = None
         while True:
             try:
-                try:
-                    q = query_queue[i].get(1, 1)
-                except TypeError:
-                    q = query_queue[i].get(1)
+                q = query_queue[i].get(timeout=1)
             except Empty:
                 q = None
             if not q:
@@ -99,10 +96,7 @@ def run_queries(i):
                     r = cursor.fetchone()
                     cursor.close()
             r = f'{i}({db._usage}): {r}'
-            try:
-                result_queue[i].put(r, 1, 1)
-            except TypeError:
-                result_queue[i].put(r, 1)
+            result_queue[i].put(r, timeout=1)
         if db:
             db.close()
 
@@ -112,68 +106,35 @@ def run_queries(i):
         threads.append(thread)
         thread.start()
     for i in range(num_threads):
-        try:
-            query_queue[i].put('ping', 1, 1)
-        except TypeError:
-            query_queue[i].put('ping', 1)
+        query_queue[i].put('ping', timeout=1)
     for i in range(num_threads):
-        try:
-            r = result_queue[i].get(1, 1)
-        except TypeError:
-            r = result_queue[i].get(1)
+        r = result_queue[i].get(timeout=1)
         assert r == f'{i}(0): ok - thread alive'
         assert threads[i].is_alive()
     for i in range(num_threads):
         for j in range(i + 1):
-            try:
-                query_queue[i].put(f'select test{j}', 1, 1)
-                r = result_queue[i].get(1, 1)
-            except TypeError:
-                query_queue[i].put(f'select test{j}', 1)
-                r = result_queue[i].get(1)
+            query_queue[i].put(f'select test{j}', timeout=1)
+            r = result_queue[i].get(timeout=1)
             assert r == f'{i}({j + 1}): test{j}'
-    try:
-        query_queue[1].put('select test4', 1, 1)
-    except TypeError:
-        query_queue[1].put('select test4', 1)
-    try:
-        r = result_queue[1].get(1, 1)
-    except TypeError:
-        r = result_queue[1].get(1)
+    query_queue[1].put('select test4', timeout=1)
+    r = result_queue[1].get(timeout=1)
     assert r == '1(3): test4'
-    try:
-        query_queue[1].put('close', 1, 1)
-        r = result_queue[1].get(1, 1)
-    except TypeError:
-        query_queue[1].put('close', 1)
-        r = result_queue[1].get(1)
+    query_queue[1].put('close', timeout=1)
+    r = result_queue[1].get(timeout=1)
     assert r == '1(3): ok - connection closed'
     for j in range(2):
-        try:
-            query_queue[1].put(f'select test{j}', 1, 1)
-            r = result_queue[1].get(1, 1)
-        except TypeError:
-            query_queue[1].put(f'select test{j}', 1)
-            r = result_queue[1].get(1)
+        query_queue[1].put(f'select test{j}', timeout=1)
+        r = result_queue[1].get(timeout=1)
         assert r == f'1({j + 1}): test{j}'
     for i in range(num_threads):
         assert threads[i].is_alive()
-        try:
-            query_queue[i].put('ping', 1, 1)
-        except TypeError:
-            query_queue[i].put('ping', 1)
+        query_queue[i].put('ping', timeout=1)
     for i in range(num_threads):
-        try:
-            r = result_queue[i].get(1, 1)
-        except TypeError:
-            r = result_queue[i].get(1)
+        r = result_queue[i].get(timeout=1)
         assert r == f'{i}({i + 1}): ok - thread alive'
         assert threads[i].is_alive()
     for i in range(num_threads):
-        try:
-            query_queue[i].put(None, 1, 1)
-        except TypeError:
-            query_queue[i].put(None, 1)
+        query_queue[i].put(None, timeout=1)
 
 
 def test_maxusage(dbapi):  # noqa: F811

tests/test_persistent_pg.py

@@ -54,10 +54,7 @@ def run_queries(i):
         db = None
         while True:
             try:
-                try:
-                    q = query_queue[i].get(1, 1)
-                except TypeError:
-                    q = query_queue[i].get(1)
+                q = query_queue[i].get(timeout=1)
             except Empty:
                 q = None
             if not q:
@@ -74,10 +71,7 @@ def run_queries(i):
                 else:
                     r = db.query(q)
             r = f'{i}({db._usage}): {r}'
-            try:
-                result_queue[i].put(r, 1, 1)
-            except TypeError:
-                result_queue[i].put(r, 1)
+            result_queue[i].put(r, timeout=1)
         if db:
             db.close()
 
@@ -87,66 +81,35 @@ def run_queries(i):
         threads.append(thread)
         thread.start()
     for i in range(num_threads):
-        try:
-            query_queue[i].put('ping', 1, 1)
-        except TypeError:
-            query_queue[i].put('ping', 1)
+        query_queue[i].put('ping', timeout=1)
     for i in range(num_threads):
-        try:
-            r = result_queue[i].get(1, 1)
-        except TypeError:
-            r = result_queue[i].get(1)
+        r = result_queue[i].get(timeout=1)
         assert r == f'{i}(0): ok - thread alive'
         assert threads[i].is_alive()
     for i in range(num_threads):
         for j in range(i + 1):
-            try:
-                query_queue[i].put(f'select test{j}', 1, 1)
-                r = result_queue[i].get(1, 1)
-            except TypeError:
-                query_queue[i].put(f'select test{j}', 1)
-                r = result_queue[i].get(1)
+            query_queue[i].put(f'select test{j}', timeout=1)
+            r = result_queue[i].get(timeout=1)
             assert r == f'{i}({j + 1}): test{j}'
-    try:
-        query_queue[1].put('select test4', 1, 1)
-        r = result_queue[1].get(1, 1)
-    except TypeError:
-        query_queue[1].put('select test4', 1)
-        r = result_queue[1].get(1)
+    query_queue[1].put('select test4', timeout=1)
+    r = result_queue[1].get(timeout=1)
     assert r == '1(3): test4'
-    try:
-        query_queue[1].put('close', 1, 1)
-        r = result_queue[1].get(1, 1)
-    except TypeError:
-        query_queue[1].put('close', 1)
-        r = result_queue[1].get(1)
+    query_queue[1].put('close', timeout=1)
+    r = result_queue[1].get(timeout=1)
     assert r == '1(3): ok - connection closed'
     for j in range(2):
-        try:
-            query_queue[1].put(f'select test{j}', 1, 1)
-            r = result_queue[1].get(1, 1)
-        except TypeError:
-            query_queue[1].put(f'select test{j}', 1)
-            r = result_queue[1].get(1)
+        query_queue[1].put(f'select test{j}', timeout=1)
+        r = result_queue[1].get(timeout=1)
         assert r == f'1({j + 1}): test{j}'
     for i in range(num_threads):
         assert threads[i].is_alive()
-        try:
-            query_queue[i].put('ping', 1, 1)
-        except TypeError:
-            query_queue[i].put('ping', 1)
+        query_queue[i].put('ping', timeout=1)
     for i in range(num_threads):
-        try:
-            r = result_queue[i].get(1, 1)
-        except TypeError:
-            r = result_queue[i].get(1)
+        r = result_queue[i].get(timeout=1)
         assert r == f'{i}({i + 1}): ok - thread alive'
         assert threads[i].is_alive()
     for i in range(num_threads):
-        try:
-            query_queue[i].put(None, 1, 1)
-        except TypeError:
-            query_queue[i].put(None, 1)
+        query_queue[i].put(None, timeout=1)
 
 
 def test_maxusage():

tests/test_pooled_db.py

@@ -1022,34 +1022,20 @@ def test_three_threads_two_connections(dbapi, threadsafety):  # noqa: F811
     queue = Queue(3)
 
     def connection():
-        try:
-            queue.put(pool.connection(), 1, 1)
-        except Exception:
-            queue.put(pool.connection(), 1)
+        queue.put(pool.connection(), timeout=1)
 
     for i in range(3):
         Thread(target=connection).start()
-    try:
-        db1 = queue.get(1, 1)
-        db2 = queue.get(1, 1)
-    except TypeError:
-        db1 = queue.get(1)
-        db2 = queue.get(1)
+    db1 = queue.get(timeout=1)
+    db2 = queue.get(timeout=1)
     assert db1 != db2
     db1_con = db1._con
     db2_con = db2._con
     assert db1_con != db2_con
-    try:
-        with pytest.raises(Empty):
-            queue.get(1, 0.1)
-    except TypeError:
-        with pytest.raises(Empty):
-            queue.get(0)
+    with pytest.raises(Empty):
+        queue.get(timeout=0.1)
     del db1
-    try:
-        db1 = queue.get(1, 1)
-    except TypeError:
-        db1 = queue.get(1)
+    db1 = queue.get(timeout=1)
     assert db1 != db2
     assert db1._con != db2._con
     assert db1._con == db1_con
@@ -1061,17 +1047,10 @@ def connection():
     db2_con = db2._con
     assert db1_con != db2_con
     Thread(target=connection).start()
-    try:
-        with pytest.raises(Empty):
-            queue.get(1, 0.1)
-    except TypeError:
-        with pytest.raises(Empty):
-            queue.get(0)
+    with pytest.raises(Empty):
+        queue.get(timeout=0.1)
     del db1
-    try:
-        db1 = queue.get(1, 1)
-    except TypeError:
-        db1 = queue.get(1)
+    db1 = queue.get(timeout=1)
     assert db1 != db2
     assert db1._con != db2._con
     assert db1._con == db1_con

tests/test_pooled_pg.py

@@ -101,13 +101,15 @@ def test_close_connection():
     db = pool.connection()
     assert pool._cache.qsize() == 1
     assert pool._cache.get(0) == db_con
+    assert db
+    del db
 
 
 def test_min_max_cached():
     pool = PooledPg(3)
     assert hasattr(pool, '_cache')
     assert pool._cache.qsize() == 3
-    cache = [pool.connection() for i in range(3)]
+    cache = [pool.connection() for _i in range(3)]
     assert pool._cache.qsize() == 0
     for i in range(3):
         cache.pop().close()
@@ -121,7 +123,7 @@ def test_min_max_cached():
     pool = PooledPg(3, 4)
     assert hasattr(pool, '_cache')
     assert pool._cache.qsize() == 3
-    cache = [pool.connection() for i in range(3)]
+    cache = [pool.connection() for _i in range(3)]
     assert pool._cache.qsize() == 0
     for i in range(3):
         cache.pop().close()
@@ -135,15 +137,15 @@ def test_min_max_cached():
     pool = PooledPg(3, 2)
     assert hasattr(pool, '_cache')
     assert pool._cache.qsize() == 3
-    cache = [pool.connection() for i in range(4)]
+    cache = [pool.connection() for _i in range(4)]
     assert pool._cache.qsize() == 0
     for i in range(4):
         cache.pop().close()
     assert pool._cache.qsize() == 3
     pool = PooledPg(2, 5)
     assert hasattr(pool, '_cache')
     assert pool._cache.qsize() == 2
-    cache = [pool.connection() for i in range(10)]
+    cache = [pool.connection() for _i in range(10)]
     assert pool._cache.qsize() == 0
     for i in range(10):
         cache.pop().close()
@@ -153,7 +155,7 @@ def test_min_max_cached():
 def test_max_connections():
     pool = PooledPg(1, 2, 3)
     assert pool._cache.qsize() == 1
-    cache = [pool.connection() for i in range(3)]
+    cache = [pool.connection() for _i in range(3)]
     assert pool._cache.qsize() == 0
     with pytest.raises(TooManyConnections):
         pool.connection()
@@ -164,7 +166,9 @@ def test_max_connections():
     assert pool._cache.qsize() == 0
     with pytest.raises(TooManyConnections):
         pool.connection()
+    assert db
     del db
+    assert cache
     del cache
     pool = PooledPg(1, 2, 1)
     assert pool._cache.qsize() == 1
@@ -176,7 +180,7 @@ def test_max_connections():
         pool.connection()
     pool = PooledPg(3, 2, 1, False)
     assert pool._cache.qsize() == 3
-    cache = [pool.connection() for i in range(3)]
+    cache = [pool.connection() for _i in range(3)]
     assert len(cache) == 3
     assert pool._cache.qsize() == 0
     with pytest.raises(TooManyConnections):
@@ -204,6 +208,7 @@ def connection():
     db = pool.connection()
     assert pool._cache.qsize() == 0
     assert session == ['thread']
+    assert db
     del db
 
 
@@ -236,34 +241,20 @@ def test_three_threads_two_connections():
     queue = Queue(3)
 
     def connection():
-        try:
-            queue.put(pool.connection(), 1, 1)
-        except TypeError:
-            queue.put(pool.connection(), 1)
+        queue.put(pool.connection(), timeout=1)
 
     for i in range(3):
         Thread(target=connection).start()
-    try:
-        db1 = queue.get(1, 1)
-        db2 = queue.get(1, 1)
-    except TypeError:
-        db1 = queue.get(1)
-        db2 = queue.get(1)
+    db1 = queue.get(timeout=1)
+    db2 = queue.get(timeout=1)
     db1_con = db1._con
     db2_con = db2._con
     assert db1 != db2
     assert db1_con != db2_con
-    try:
-        with pytest.raises(Empty):
-            queue.get(1, 0.1)
-    except TypeError:
-        with pytest.raises(Empty):
-            queue.get(0)
+    with pytest.raises(Empty):
+        queue.get(timeout=0.1)
     del db1
-    try:
-        db1 = queue.get(1, 1)
-    except TypeError:
-        db1 = queue.get(1)
+    db1 = queue.get(timeout=1)
     assert db1 != db2
     assert db1._con != db2._con
     assert db1._con == db1_con