Apply ruff linting and formatting fixes

Co-authored-by: jason810496 <[email protected]>

Author: Copilot

README.md

@@ -119,7 +119,7 @@ See [`Development` on Online Document](https://pgmq-sqlalchemy.readthedocs.io/en
 
 ## TODO 
 
-- [ ] Add **time-based** partition option and validation to `create_partitioned_queue` method.
-- [ ] Read(single/batch) Archive Table ( `read_archive` method )
-- [ ] Detach Archive Table ( `detach_archive` method )
-- [ ] Add `set_vt` utils method.
+- [x] Add **time-based** partition option and validation to `create_partitioned_queue` method.
+- [x] Read(single/batch) Archive Table ( `read_archive` method )
+- [x] Detach Archive Table ( `detach_archive` method )
+- [x] Add `set_vt` utils method.

pgmq_sqlalchemy/queue.py

@@ -242,25 +242,25 @@ async def _create_partitioned_queue_async(
 
     def _validate_partition_interval(self, interval: Union[int, str]) -> str:
         """Validate partition interval format.
-        
+
         Args:
             interval: Either an integer for numeric partitioning or a string for time-based partitioning
                      (e.g., '1 day', '1 hour', '7 days')
-        
+
         Returns:
             The validated interval as a string
-            
+
         Raises:
             ValueError: If the interval format is invalid
         """
         if isinstance(interval, int):
             if interval <= 0:
                 raise ValueError("Numeric partition interval must be positive")
             return str(interval)
-        
+
         # Validate time-based interval format
         # Valid PostgreSQL interval formats: '1 day', '7 days', '1 hour', '1 month', etc.
-        time_pattern = r'^\d+\s+(microsecond|millisecond|second|minute|hour|day|week|month|year)s?$'
+        time_pattern = r"^\d+\s+(microsecond|millisecond|second|minute|hour|day|week|month|year)s?$"
         if not re.match(time_pattern, interval.strip(), re.IGNORECASE):
             raise ValueError(
                 f"Invalid time-based partition interval: '{interval}'. "
@@ -284,7 +284,7 @@ def create_partitioned_queue(
 
                 # Numeric partitioning (by msg_id)
                 pgmq_client.create_partitioned_queue('my_partitioned_queue', partition_interval=10000, retention_interval=100000)
-                
+
                 # Time-based partitioning (by enqueued_at)
                 pgmq_client.create_partitioned_queue('my_time_queue', partition_interval='1 day', retention_interval='7 days')
 
@@ -309,15 +309,21 @@ def create_partitioned_queue(
         """
         # check if the pg_partman extension exists before creating a partitioned queue at runtime
         self._check_pg_partman_ext()
-        
+
         # Validate partition intervals
-        validated_partition_interval = self._validate_partition_interval(partition_interval)
-        validated_retention_interval = self._validate_partition_interval(retention_interval)
+        validated_partition_interval = self._validate_partition_interval(
+            partition_interval
+        )
+        validated_retention_interval = self._validate_partition_interval(
+            retention_interval
+        )
 
         if self.is_async:
             return self.loop.run_until_complete(
                 self._create_partitioned_queue_async(
-                    queue_name, validated_partition_interval, validated_retention_interval
+                    queue_name,
+                    validated_partition_interval,
+                    validated_retention_interval,
                 )
             )
         return self._create_partitioned_queue_sync(
@@ -1460,7 +1466,9 @@ def _read_archive_sync(self, queue_name: str) -> Optional[Message]:
         """Read a single message from the archive table synchronously."""
         with self.session_maker() as session:
             row = session.execute(
-                text(f"select msg_id, read_ct, enqueued_at, vt, message from pgmq.a_{queue_name} limit 1;")
+                text(
+                    f"select msg_id, read_ct, enqueued_at, vt, message from pgmq.a_{queue_name} limit 1;"
+                )
             ).fetchone()
             session.commit()
         if row is None:
@@ -1474,7 +1482,9 @@ async def _read_archive_async(self, queue_name: str) -> Optional[Message]:
         async with self.session_maker() as session:
             row = (
                 await session.execute(
-                    text(f"select msg_id, read_ct, enqueued_at, vt, message from pgmq.a_{queue_name} limit 1;")
+                    text(
+                        f"select msg_id, read_ct, enqueued_at, vt, message from pgmq.a_{queue_name} limit 1;"
+                    )
                 )
             ).fetchone()
             await session.commit()

tests/test_queue.py

@@ -459,35 +459,44 @@ def test_detach_archive(pgmq_fixture, db_session):
     msg = MSG
     msg_id = pgmq.send(queue_name, msg)
     pgmq.archive(queue_name, msg_id)
-    
+
     # Detach archive should not raise an error
     pgmq.detach_archive(queue_name)
-    
+
     # Read the archive to ensure it still exists after detaching
     archived_msg = pgmq.read_archive(queue_name)
     assert archived_msg is not None
     assert archived_msg.msg_id == msg_id
-    
+
     # Cleanup: Drop the archive and queue tables
     # After detaching, the archive is no longer part of the extension
     # We need to drop both tables manually by first removing them from the extension
     if pgmq.is_async:
-        import asyncio
+
         async def cleanup():
             async with pgmq.session_maker() as session:
                 # Drop archive table (already detached)
-                await session.execute(text(f"DROP TABLE IF EXISTS pgmq.a_{queue_name} CASCADE;"))
+                await session.execute(
+                    text(f"DROP TABLE IF EXISTS pgmq.a_{queue_name} CASCADE;")
+                )
                 # Detach and drop queue table
-                await session.execute(text(f"ALTER EXTENSION pgmq DROP TABLE pgmq.q_{queue_name};"))
-                await session.execute(text(f"DROP TABLE IF EXISTS pgmq.q_{queue_name} CASCADE;"))
+                await session.execute(
+                    text(f"ALTER EXTENSION pgmq DROP TABLE pgmq.q_{queue_name};")
+                )
+                await session.execute(
+                    text(f"DROP TABLE IF EXISTS pgmq.q_{queue_name} CASCADE;")
+                )
                 await session.commit()
+
         pgmq.loop.run_until_complete(cleanup())
     else:
         with pgmq.session_maker() as session:
             # Drop archive table (already detached)
             session.execute(text(f"DROP TABLE IF EXISTS pgmq.a_{queue_name} CASCADE;"))
             # Detach and drop queue table
-            session.execute(text(f"ALTER EXTENSION pgmq DROP TABLE pgmq.q_{queue_name};"))
+            session.execute(
+                text(f"ALTER EXTENSION pgmq DROP TABLE pgmq.q_{queue_name};")
+            )
             session.execute(text(f"DROP TABLE IF EXISTS pgmq.q_{queue_name} CASCADE;"))
             session.commit()
 
@@ -544,22 +553,30 @@ def test_read_archive_batch_limit(pgmq_setup_teardown: PGMQ_WITH_QUEUE):
 def test_create_time_based_partitioned_queue(pgmq_fixture, db_session):
     pgmq: PGMQueue = pgmq_fixture
     queue_name = f"test_queue_{uuid.uuid4().hex}"
-    pgmq.create_partitioned_queue(queue_name, partition_interval='1 day', retention_interval='7 days')
+    pgmq.create_partitioned_queue(
+        queue_name, partition_interval="1 day", retention_interval="7 days"
+    )
     assert check_queue_exists(db_session, queue_name) is True
 
 
 @pgmq_deps
-def test_create_time_based_partitioned_queue_various_intervals(pgmq_fixture, db_session):
+def test_create_time_based_partitioned_queue_various_intervals(
+    pgmq_fixture, db_session
+):
     pgmq: PGMQueue = pgmq_fixture
-    
+
     # Test with hour
     queue_name_hour = f"test_queue_{uuid.uuid4().hex}"
-    pgmq.create_partitioned_queue(queue_name_hour, partition_interval='1 hour', retention_interval='24 hours')
+    pgmq.create_partitioned_queue(
+        queue_name_hour, partition_interval="1 hour", retention_interval="24 hours"
+    )
     assert check_queue_exists(db_session, queue_name_hour) is True
-    
+
     # Test with week
     queue_name_week = f"test_queue_{uuid.uuid4().hex}"
-    pgmq.create_partitioned_queue(queue_name_week, partition_interval='1 week', retention_interval='4 weeks')
+    pgmq.create_partitioned_queue(
+        queue_name_week, partition_interval="1 week", retention_interval="4 weeks"
+    )
     assert check_queue_exists(db_session, queue_name_week) is True
 
 
@@ -568,7 +585,11 @@ def test_create_partitioned_queue_invalid_time_interval(pgmq_fixture):
     pgmq: PGMQueue = pgmq_fixture
     queue_name = f"test_queue_{uuid.uuid4().hex}"
     with pytest.raises(ValueError) as e:
-        pgmq.create_partitioned_queue(queue_name, partition_interval='invalid interval', retention_interval='7 days')
+        pgmq.create_partitioned_queue(
+            queue_name,
+            partition_interval="invalid interval",
+            retention_interval="7 days",
+        )
     assert "Invalid time-based partition interval" in str(e.value)
 
 
@@ -577,5 +598,7 @@ def test_create_partitioned_queue_invalid_numeric_interval(pgmq_fixture):
     pgmq: PGMQueue = pgmq_fixture
     queue_name = f"test_queue_{uuid.uuid4().hex}"
     with pytest.raises(ValueError) as e:
-        pgmq.create_partitioned_queue(queue_name, partition_interval=-100, retention_interval=100000)
+        pgmq.create_partitioned_queue(
+            queue_name, partition_interval=-100, retention_interval=100000
+        )
     assert "Numeric partition interval must be positive" in str(e.value)