index

async_pq/queue.py

@@ -4,7 +4,7 @@
 
 from asyncpg import Connection
 
-DELETE_LIMIT_SM = 10
+DELETE_LIMIT_SM = 1000  # limit of requests deleted in one query
 
 
 class Queue:
@@ -93,8 +93,8 @@ async def return_unacked(self, timeout: int, limit: int=DELETE_LIMIT_SM) -> int:
             f"""
             WITH deleted AS (
               DELETE FROM {self._requests_table_name} 
-              WHERE ctid IN (
-                SELECT ctid
+              WHERE r_id IN (
+                SELECT r_id
                 FROM {self._requests_table_name}
                 WHERE r_status='wait' AND created_at < current_timestamp - $1::interval
                 LIMIT $2
@@ -161,6 +161,7 @@ async def _new_queue(self, name: str):
               q_data JSON,
               q_request_id INT REFERENCES queue_{name}_requests(r_id) ON DELETE SET NULL
             );
+            CREATE INDEX queue_{name}_q_request_id_idx ON queue_{name} (q_request_id);
             """
         )
 

docs/README.md

@@ -50,12 +50,14 @@ Or vice versa:
 is_unacked: bool = await queue.unack(request_id)
 ```
 
-Return to queue all unacknowledged massages older than ```timeout``` seconds:
+Return to queue all unacknowledged massages older than ```timeout``` seconds 
+(default limit=1000 entities):
 ```python
 await queue.return_unacked(timeout=300)
 ```
 
-Clean queue (delete all acknowledged massages) to not overfill database with old data:
+Clean queue (delete acknowledged massages) to not overfill database with old data 
+(default limit=1000 entities):
 ```python
 await queue.clean_acked_queue()
 ```

setup.py

@@ -10,9 +10,9 @@ def readme(fname):
 
 setup(
     name='async-pq',
-    version='0.2.2',
+    version='0.3.0',
     description='Python async api for creating and managing queues in postgres',
-    long_description=readme('README.md'),
+    long_description=readme('./docs/README.md'),
     long_description_content_type="text/markdown",
     author='Danilchenko Maksim',
     author_email='dmax.dev@gmail.com',