Move _validate_partition_interval to operations

Author: jason810496

pgmq_sqlalchemy/operation.py

@@ -1,4 +1,5 @@
-from typing import List, Optional, Tuple, Dict, Any
+from typing import List, Optional, Tuple, Dict, Any, Union
+import re
 
 from sqlalchemy import text
 from sqlalchemy.orm import Session
@@ -28,6 +29,36 @@ def _get_check_pg_partman_ext_statement() -> Tuple[str, Dict[str, Any]]:
         """Get statement and params for checking/creating pg_partman extension."""
         return "create extension if not exists pg_partman cascade;", {}
 
+    @staticmethod
+    def _validate_partition_interval(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?$"
+        if not re.match(time_pattern, interval.strip(), re.IGNORECASE):
+            raise ValueError(
+                f"Invalid time-based partition interval: '{interval}'. "
+                "Expected format: '<number> <unit>' where unit is one of: "
+                "microsecond, millisecond, second, minute, hour, day, week, month, year"
+            )
+        return interval.strip()
+
     @staticmethod
     def _get_create_queue_statement(
         queue_name: str, unlogged: bool
@@ -349,6 +380,14 @@ async def create_partitioned_queue_async(
             session: Async SQLAlchemy session.
             commit: Whether to commit the transaction.
         """
+        # Validate partition intervals
+        partition_interval = PGMQOperation._validate_partition_interval(
+            partition_interval
+        )
+        retention_interval = PGMQOperation._validate_partition_interval(
+            retention_interval
+        )
+
         stmt, params = PGMQOperation._get_create_partitioned_queue_statement(
             queue_name, partition_interval, retention_interval
         )

pgmq_sqlalchemy/queue.py

@@ -1,6 +1,5 @@
 import asyncio
-import re
-from typing import List, Optional, Union
+from typing import List, Optional
 
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
@@ -191,35 +190,6 @@ async def _create():
             queue_name, unlogged, session=session, commit=commit
         )
 
-    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?$"
-        if not re.match(time_pattern, interval.strip(), re.IGNORECASE):
-            raise ValueError(
-                f"Invalid time-based partition interval: '{interval}'. "
-                "Expected format: '<number> <unit>' where unit is one of: "
-                "microsecond, millisecond, second, minute, hour, day, week, month, year"
-            )
-        return interval.strip()
-
     def create_partitioned_queue(
         self,
         queue_name: str,
@@ -264,10 +234,6 @@ 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
-        partition_interval = self._validate_partition_interval(partition_interval)
-        retention_interval = self._validate_partition_interval(retention_interval)
-
         if self.is_async:
             if session is None: