feat: implement pagination bounds validation to criteria converters and URL converter

Signed-off-by: Adria Montoto <75563346+adriamontoto@users.noreply.github.com>

Author: adriamontoto

criteria_pattern/converters/criteria_to_mysql_converter.py

@@ -6,7 +6,13 @@
 from typing import Any, assert_never
 
 from criteria_pattern import Criteria, Direction, Operator
-from criteria_pattern.errors import InvalidColumnError, InvalidDirectionError, InvalidOperatorError, InvalidTableError
+from criteria_pattern.errors import (
+    InvalidColumnError,
+    InvalidDirectionError,
+    InvalidOperatorError,
+    InvalidTableError,
+    PaginationBoundsError,
+)
 from criteria_pattern.models.criteria import AndCriteria, NotCriteria, OrCriteria
 
 
@@ -32,7 +38,7 @@ class CriteriaToMysqlConverter:
     """  # noqa: E501  # fmt: skip
 
     @classmethod
-    def convert(
+    def convert(  # noqa: C901
         cls,
         criteria: Criteria,
         table: str,
@@ -43,10 +49,13 @@ def convert(
         check_criteria_injection: bool = False,
         check_operator_injection: bool = False,
         check_direction_injection: bool = False,
+        check_pagination_bounds: bool = False,
         valid_tables: Sequence[str] | None = None,
         valid_columns: Sequence[str] | None = None,
         valid_operators: Sequence[Operator] | None = None,
         valid_directions: Sequence[Direction] | None = None,
+        max_page_size: int = 10000,
+        max_page_number: int = 1000000,
     ) -> tuple[str, list[Any]]:
         """
         Convert the Criteria object to a MySQL query.
@@ -66,16 +75,21 @@ def convert(
             operators. Default to False.
             check_direction_injection (bool, optional): Raise an error if the direction is not in the list of valid
             directions. Default to False.
+            check_pagination_bounds (bool, optional): Raise an error if pagination parameters exceed maximum bounds.
+            Default to False.
             valid_tables (Sequence[str], optional): List of valid tables to query. Default to empty list.
             valid_columns (Sequence[str], optional): List of valid columns to select. Default to empty list.
             valid_operators (Sequence[Operator], optional): List of valid operators to use. Default to empty list.
             valid_directions (Sequence[Direction], optional): List of valid directions to use. Default to empty list.
+            max_page_size (int, optional): Maximum allowed page_size to prevent integer overflow. Default to 10000.
+            max_page_number (int, optional): Maximum allowed page_number to prevent integer overflow. Default to 1000000.
 
         Raises:
             InvalidTableError: If the table is not in the list of valid tables (only if check_table_injection=True).
             InvalidColumnError: If the column is not in the list of valid columns (only if check_column_injection=True).
             InvalidOperatorError: If the operator is not in the list of valid operators (only if check_operator_injection=True).
             InvalidDirectionError: If the direction is not in the list of valid directions (only if check_direction_injection=True).
+            PaginationBoundsError: If pagination parameters exceed maximum bounds (only if check_pagination_bounds=True).
 
         Returns:
             tuple[str, list[Any]]: The MySQL query string and the query parameters as a list.
@@ -118,6 +132,13 @@ def convert(
         if check_direction_injection:
             cls._validate_directions(criteria=criteria, valid_directions=valid_directions)
 
+        if check_pagination_bounds:
+            cls._validate_pagination_bounds(
+                criteria=criteria,
+                max_page_size=max_page_size,
+                max_page_number=max_page_number,
+            )
+
         query = f'SELECT {", ".join(columns)} FROM {table}'  # noqa: S608  # nosec
         parameters: list[Any] = []
         parameters_counter = 0
@@ -241,6 +262,25 @@ def _validate_directions(cls, *, criteria: Criteria, valid_directions: Sequence[
                     valid_directions=valid_directions,
                 )
 
+    @classmethod
+    def _validate_pagination_bounds(cls, *, criteria: Criteria, max_page_size: int, max_page_number: int) -> None:
+        """
+        Validate the Criteria object pagination parameters to prevent integer overflow.
+
+        Args:
+            criteria (Criteria): Criteria to validate.
+            max_page_size (int): Maximum allowed page_size.
+            max_page_number (int): Maximum allowed page_number.
+
+        Raises:
+            PaginationBoundsError: If pagination parameters exceed maximum bounds.
+        """
+        if criteria.page_size is not None and criteria.page_size > max_page_size:
+            raise PaginationBoundsError(parameter='page_size', value=criteria.page_size, max_value=max_page_size)
+
+        if criteria.page_number is not None and criteria.page_number > max_page_number:
+            raise PaginationBoundsError(parameter='page_number', value=criteria.page_number, max_value=max_page_number)
+
     @classmethod
     def _process_filters(cls, *, criteria: Criteria, columns_mapping: Mapping[str, str]) -> tuple[str, list[Any]]:
         """

criteria_pattern/converters/criteria_to_postgresql_converter.py

@@ -6,7 +6,13 @@
 from typing import Any, assert_never
 
 from criteria_pattern import Criteria, Direction, Operator
-from criteria_pattern.errors import InvalidColumnError, InvalidDirectionError, InvalidOperatorError, InvalidTableError
+from criteria_pattern.errors import (
+    InvalidColumnError,
+    InvalidDirectionError,
+    InvalidOperatorError,
+    InvalidTableError,
+    PaginationBoundsError,
+)
 from criteria_pattern.models.criteria import AndCriteria, NotCriteria, OrCriteria
 
 
@@ -32,7 +38,7 @@ class CriteriaToPostgresqlConverter:
     """  # noqa: E501  # fmt: skip
 
     @classmethod
-    def convert(
+    def convert(  # noqa: C901
         cls,
         criteria: Criteria,
         table: str,
@@ -43,10 +49,13 @@ def convert(
         check_criteria_injection: bool = False,
         check_operator_injection: bool = False,
         check_direction_injection: bool = False,
+        check_pagination_bounds: bool = False,
         valid_tables: Sequence[str] | None = None,
         valid_columns: Sequence[str] | None = None,
         valid_operators: Sequence[Operator] | None = None,
         valid_directions: Sequence[Direction] | None = None,
+        max_page_size: int = 10000,
+        max_page_number: int = 1000000,
     ) -> tuple[str, dict[str, Any]]:
         """
         Convert the Criteria object to a Postgresql query.
@@ -66,16 +75,21 @@ def convert(
             Default to False.
             check_direction_injection (bool, optional): Raise an error if the direction is not in the list of valid
             directions. Default to False.
+            check_pagination_bounds (bool, optional): Raise an error if pagination parameters exceed maximum bounds.
+            Default to False.
             valid_tables (Sequence[str], optional): List of valid tables to query. Default to empty list.
             valid_columns (Sequence[str], optional): List of valid columns to select. Default to empty list.
             valid_operators (Sequence[Operator], optional): List of valid operators to use. Default to empty list.
             valid_directions (Sequence[Direction], optional): List of valid directions to use. Default to empty list.
+            max_page_size (int, optional): Maximum allowed page_size to prevent integer overflow. Default to 10000.
+            max_page_number (int, optional): Maximum allowed page_number to prevent integer overflow. Default to 1000000.
 
         Raises:
             InvalidTableError: If the table is not in the list of valid tables (only if check_table_injection=True).
             InvalidColumnError: If the column is not in the list of valid columns (only if check_column_injection=True).
             InvalidOperatorError: If the operator is not in the list of valid operators (only if check_operator_injection=True).
             InvalidDirectionError: If the direction is not in the list of valid directions (only if check_direction_injection=True).
+            PaginationBoundsError: If pagination parameters exceed maximum bounds (only if check_pagination_bounds=True).
 
         Returns:
             tuple[str, dict[str, Any]]: The Postgresql query string and the query parameters.
@@ -118,6 +132,13 @@ def convert(
         if check_direction_injection:
             cls._validate_directions(criteria=criteria, valid_directions=valid_directions)
 
+        if check_pagination_bounds:
+            cls._validate_pagination_bounds(
+                criteria=criteria,
+                max_page_size=max_page_size,
+                max_page_number=max_page_number,
+            )
+
         quoted_columns = ['*' if column == '*' else f'"{column}"' for column in columns]
         quoted_table = '.'.join(f'"{part}"' for part in table.split('.'))
         query = f'SELECT {", ".join(quoted_columns)} FROM {quoted_table}'  # noqa: S608  # nosec
@@ -245,6 +266,25 @@ def _validate_directions(cls, *, criteria: Criteria, valid_directions: Sequence[
                     valid_directions=valid_directions,
                 )
 
+    @classmethod
+    def _validate_pagination_bounds(cls, *, criteria: Criteria, max_page_size: int, max_page_number: int) -> None:
+        """
+        Validate the Criteria object pagination parameters to prevent integer overflow.
+
+        Args:
+            criteria (Criteria): Criteria to validate.
+            max_page_size (int): Maximum allowed page_size.
+            max_page_number (int): Maximum allowed page_number.
+
+        Raises:
+            PaginationBoundsError: If pagination parameters exceed maximum bounds.
+        """
+        if criteria.page_size is not None and criteria.page_size > max_page_size:
+            raise PaginationBoundsError(parameter='page_size', value=criteria.page_size, max_value=max_page_size)
+
+        if criteria.page_number is not None and criteria.page_number > max_page_number:
+            raise PaginationBoundsError(parameter='page_number', value=criteria.page_number, max_value=max_page_number)
+
     @classmethod
     def _process_filters(cls, *, criteria: Criteria, columns_mapping: Mapping[str, str]) -> tuple[str, dict[str, Any]]:
         """

criteria_pattern/converters/criteria_to_sqlite_converter.py

@@ -6,7 +6,13 @@
 from typing import Any, assert_never
 
 from criteria_pattern import Criteria, Direction, Operator
-from criteria_pattern.errors import InvalidColumnError, InvalidDirectionError, InvalidOperatorError, InvalidTableError
+from criteria_pattern.errors import (
+    InvalidColumnError,
+    InvalidDirectionError,
+    InvalidOperatorError,
+    InvalidTableError,
+    PaginationBoundsError,
+)
 from criteria_pattern.models.criteria import AndCriteria, NotCriteria, OrCriteria
 
 
@@ -32,7 +38,7 @@ class CriteriaToSqliteConverter:
     """  # noqa: E501  # fmt: skip
 
     @classmethod
-    def convert(
+    def convert(  # noqa: C901
         cls,
         criteria: Criteria,
         table: str,
@@ -43,10 +49,13 @@ def convert(
         check_criteria_injection: bool = False,
         check_operator_injection: bool = False,
         check_direction_injection: bool = False,
+        check_pagination_bounds: bool = False,
         valid_tables: Sequence[str] | None = None,
         valid_columns: Sequence[str] | None = None,
         valid_operators: Sequence[Operator] | None = None,
         valid_directions: Sequence[Direction] | None = None,
+        max_page_size: int = 10000,
+        max_page_number: int = 1000000,
     ) -> tuple[str, dict[str, Any]]:
         """
         Convert the Criteria object to a SQLite query.
@@ -66,16 +75,21 @@ def convert(
             operators. Default to False.
             check_direction_injection (bool, optional): Raise an error if the direction is not in the list of valid
             directions. Default to False.
+            check_pagination_bounds (bool, optional): Raise an error if pagination parameters exceed maximum bounds.
+            Default to False.
             valid_tables (Sequence[str], optional): List of valid tables to query. Default to empty list.
             valid_columns (Sequence[str], optional): List of valid columns to select. Default to empty list.
             valid_operators (Sequence[Operator], optional): List of valid operators to use. Default to empty list.
             valid_directions (Sequence[Direction], optional): List of valid directions to use. Default to empty list.
+            max_page_size (int, optional): Maximum allowed page_size to prevent integer overflow. Default to 10000.
+            max_page_number (int, optional): Maximum allowed page_number to prevent integer overflow. Default to 1000000.
 
         Raises:
             InvalidTableError: If the table is not in the list of valid tables (only if check_table_injection=True).
             InvalidColumnError: If the column is not in the list of valid columns (only if check_column_injection=True).
             InvalidOperatorError: If the operator is not in the list of valid operators (only if check_operator_injection=True).
             InvalidDirectionError: If the direction is not in the list of valid directions (only if check_direction_injection=True).
+            PaginationBoundsError: If pagination parameters exceed maximum bounds (only if check_pagination_bounds=True).
 
         Returns:
             tuple[str, dict[str, Any]]: The SQLite query string and the query parameters.
@@ -118,6 +132,13 @@ def convert(
         if check_direction_injection:
             cls._validate_directions(criteria=criteria, valid_directions=valid_directions)
 
+        if check_pagination_bounds:
+            cls._validate_pagination_bounds(
+                criteria=criteria,
+                max_page_size=max_page_size,
+                max_page_number=max_page_number,
+            )
+
         quoted_columns = ['*' if column == '*' else f'"{column}"' for column in columns]
         quoted_table = '.'.join(f'"{part}"' for part in table.split('.'))
         query = f'SELECT {", ".join(quoted_columns)} FROM {quoted_table}'  # noqa: S608  # nosec
@@ -245,6 +266,25 @@ def _validate_directions(cls, *, criteria: Criteria, valid_directions: Sequence[
                     valid_directions=valid_directions,
                 )
 
+    @classmethod
+    def _validate_pagination_bounds(cls, *, criteria: Criteria, max_page_size: int, max_page_number: int) -> None:
+        """
+        Validate the Criteria object pagination parameters to prevent integer overflow.
+
+        Args:
+            criteria (Criteria): Criteria to validate.
+            max_page_size (int): Maximum allowed page_size.
+            max_page_number (int): Maximum allowed page_number.
+
+        Raises:
+            PaginationBoundsError: If pagination parameters exceed maximum bounds.
+        """
+        if criteria.page_size is not None and criteria.page_size > max_page_size:
+            raise PaginationBoundsError(parameter='page_size', value=criteria.page_size, max_value=max_page_size)
+
+        if criteria.page_number is not None and criteria.page_number > max_page_number:
+            raise PaginationBoundsError(parameter='page_number', value=criteria.page_number, max_value=max_page_number)
+
     @classmethod
     def _process_filters(cls, *, criteria: Criteria, columns_mapping: Mapping[str, str]) -> tuple[str, dict[str, Any]]:
         """

criteria_pattern/converters/url_to_criteria_converter.py

@@ -8,7 +8,12 @@
 from urllib.parse import parse_qs, unquote_plus, urlparse
 
 from criteria_pattern import Criteria, Direction, Filter, Operator, Order
-from criteria_pattern.errors import InvalidColumnError, InvalidDirectionError, InvalidOperatorError
+from criteria_pattern.errors import (
+    InvalidColumnError,
+    InvalidDirectionError,
+    InvalidOperatorError,
+    PaginationBoundsError,
+)
 
 
 class UrlToCriteriaConverter:
@@ -67,9 +72,12 @@ def convert(
         check_field_injection: bool = False,
         check_operator_injection: bool = False,
         check_direction_injection: bool = False,
+        check_pagination_bounds: bool = False,
         valid_fields: Sequence[str] | None = None,
         valid_operators: Sequence[Operator] | None = None,
         valid_directions: Sequence[Direction] | None = None,
+        max_page_size: int = 10000,
+        max_page_number: int = 1000000,
     ) -> Criteria:
         """
         Converts an URL query string into a Criteria object.
@@ -80,9 +88,12 @@ def convert(
             check_field_injection (bool, optional): Whether to check for field injection.
             check_operator_injection (bool, optional): Whether to check for operator injection.
             check_direction_injection (bool, optional): Whether to check for direction injection.
+            check_pagination_bounds (bool, optional): Whether to check pagination parameters bounds.
             valid_fields (Sequence[str], optional): A list of valid field names. Default to empty list.
             valid_operators (Sequence[Operator], optional): A list of valid operators. Default to empty list.
             valid_directions (Sequence[Direction], optional): A list of valid directions. Default to empty list.
+            max_page_size (int, optional): Maximum allowed page_size to prevent integer overflow. Default to 10000.
+            max_page_number (int, optional): Maximum allowed page_number to prevent integer overflow. Default to 1000000.
 
         Raises:
             TypeError: If the filter index is not an integer.
@@ -98,6 +109,7 @@ def convert(
             InvalidColumnError: If an invalid field name is found in orders.
             InvalidOperatorError: If an invalid operator is found in filters.
             InvalidDirectionError: If an invalid direction is found in orders.
+            PaginationBoundsError: If pagination parameters exceed maximum bounds.
 
         Example:
         ```python
@@ -137,6 +149,13 @@ def convert(
         if check_direction_injection:
             cls._validate_directions(criteria=criteria, valid_directions=valid_directions)
 
+        if check_pagination_bounds:
+            cls._validate_pagination_bounds(
+                criteria=criteria,
+                max_page_size=max_page_size,
+                max_page_number=max_page_number,
+            )
+
         return criteria
 
     @classmethod
@@ -445,3 +464,22 @@ def _validate_directions(cls, *, criteria: Criteria, valid_directions: Sequence[
                     direction=Direction(value=order.direction),
                     valid_directions=valid_directions,
                 )
+
+    @classmethod
+    def _validate_pagination_bounds(cls, *, criteria: Criteria, max_page_size: int, max_page_number: int) -> None:
+        """
+        Validate the Criteria object pagination parameters to prevent integer overflow.
+
+        Args:
+            criteria (Criteria): Criteria to validate.
+            max_page_size (int): Maximum allowed page_size.
+            max_page_number (int): Maximum allowed page_number.
+
+        Raises:
+            PaginationBoundsError: If pagination parameters exceed maximum bounds.
+        """
+        if criteria.page_size is not None and criteria.page_size > max_page_size:
+            raise PaginationBoundsError(parameter='page_size', value=criteria.page_size, max_value=max_page_size)
+
+        if criteria.page_number is not None and criteria.page_number > max_page_number:
+            raise PaginationBoundsError(parameter='page_number', value=criteria.page_number, max_value=max_page_number)

criteria_pattern/errors/__init__.py

@@ -2,12 +2,14 @@
 from .invalid_direction_error import InvalidDirectionError
 from .invalid_operator_error import InvalidOperatorError
 from .invalid_table_error import InvalidTableError
+from .pagination_bounds_error import PaginationBoundsError
 from .sql_converter_error import SqlConverterError
 
 __all__ = (
     'InvalidColumnError',
     'InvalidDirectionError',
     'InvalidOperatorError',
     'InvalidTableError',
+    'PaginationBoundsError',
     'SqlConverterError',
 )