Repository: igorbenav/fastcrud
Pull request #210 (2025-03-21 17:19:46)

Implement OR and NOT for filtering.
# Pull Request Template for FastCRUD

## Description
refactor: improve FastCRUD filter parsing readability and operations
- Split _parse_filters into smaller, focused helper methods
- Added proper type hints and docstrings for better clarity
- Fixed variable shadowing in column handling
- Improved error handling consistency
- Added dedicated methods for:
  - _handle_simple_filter
  - _handle_or_filter
  - _handle_not_filter
  - _handle_standard_filter
  - _get_column

Implementation Notes:
- OR operations: field__or={'gt': 18, 'lt': 65} applies multiple conditions to same field Example: age__or={'gt': 18, 'lt': 65} -> WHERE age > 18 OR age < 65

- NOT operations: field__not={'eq': value, 'between': (x,y)} applies NOT to each condition
  Example: age__not={'eq': 20, 'between': (30,40)} ->
  WHERE NOT age = 20 AND NOT (age BETWEEN 30 AND 40)

Makes the code more maintainable and easier to test by reducing complexity and improving separation of concerns.

## Changes
Briefly list the changes you've made. If applicable, also link any relevant issues or pull requests.

## Tests
Describe the tests you added or modified to cover your changes, if applicable.

## Checklist
- [x] I have read the [CONTRIBUTING](CONTRIBUTING.md) document.
- [x] My code follows the code style of this project.
- [x] I have added necessary documentation (if appropriate).
- [x] I have added tests that cover my changes (if applicable).
- [x] All new and existing tests passed.

## Additional Notes



doubledare704: Related to 
[Issue1](https://github.com/igorbenav/fastcrud/issues/205)
[Issue2](https://github.com/igorbenav/fastcrud/issues/206)
igorbenav: @doubledare704  can you please just fix the typing issues?

Patch:
diff --git a/docs/advanced/filters.md b/docs/advanced/filters.md
new file mode 100644
index 0000000..d714c67
--- /dev/null
+++ b/docs/advanced/filters.md
@@ -0,0 +1,86 @@
+# Advanced Filtering
+
+The `_parse_filters` method in FastCRUD supports complex filtering operations including OR and NOT conditions.
+
+## Basic Usage
+
+Filters are specified as keyword arguments in the format `field_name__operator=value`:
+
+```python
+# Simple equality filter
+results = await crud.get_multi(db, name="John")
+
+# Comparison operators
+results = await crud.get_multi(db, age__gt=18)
+```
+
+## OR Operations
+
+Use the `__or` suffix to apply multiple conditions to the same field with OR logic:
+
+```python
+# Find users aged under 18 OR over 65
+results = await crud.get_multi(
+    db,
+    age__or={
+        "lt": 18,
+        "gt": 65
+    }
+)
+# Generates: WHERE age < 18 OR age > 65
+```
+
+## NOT Operations
+
+Use the `__not` suffix to negate multiple conditions on the same field:
+
+```python
+# Find users NOT aged 20 AND NOT between 30-40
+results = await crud.get_multi(
+    db,
+    age__not={
+        "eq": 20,
+        "between": (30, 40)
+    }
+)
+# Generates: WHERE NOT age = 20 AND NOT (age BETWEEN 30 AND 40)
+```
+
+## Supported Operators
+
+- Comparison: `eq`, `gt`, `lt`, `gte`, `lte`, `ne`
+- Null checks: `is`, `is_not`
+- Text matching: `like`, `notlike`, `ilike`, `notilike`, `startswith`, `endswith`, `contains`, `match`
+- Collections: `in`, `not_in`, `between`
+- Logical: `or`, `not`
+
+## Examples
+
+```python
+# Complex age filtering
+results = await crud.get_multi(
+    db,
+    age__or={
+        "between": (20, 30),
+        "eq": 18
+    },
+    status__not={
+        "in": ["inactive", "banned"]
+    }
+)
+
+# Text search with OR conditions
+results = await crud.get_multi(
+    db,
+    name__or={
+        "startswith": "A",
+        "endswith": "smith"
+    }
+)
+```
+
+## Error Handling
+
+- Invalid column names raise `ValueError`
+- Invalid operators are ignored
+- Invalid value types for operators (e.g., non-list for `between`) raise `ValueError`
\ No newline at end of file
diff --git a/fastcrud/crud/fast_crud.py b/fastcrud/crud/fast_crud.py
index 5c141df..5a61974 100644
--- a/fastcrud/crud/fast_crud.py
+++ b/fastcrud/crud/fast_crud.py
@@ -1,4 +1,4 @@
-from typing import Any, Generic, Union, Optional, Callable
+from typing import Any, Generic, Union, Optional, Callable, cast
 from datetime import datetime, timezone
 
 from pydantic import ValidationError
@@ -15,6 +15,8 @@ from sqlalchemy import (
     desc,
     or_,
     column,
+    not_,
+    Column,
 )
 from sqlalchemy.exc import ArgumentError, MultipleResultsFound, NoResultFound
 from sqlalchemy.sql import Join
@@ -47,6 +49,7 @@ from .helper import (
 
 from ..endpoint.helper import _get_primary_keys
 
+FilterCallable = Callable[[Column[Any]], Callable[..., ColumnElement[bool]]]
 
 class FastCRUD(
     Generic[
@@ -453,6 +456,7 @@ class FastCRUD(
     """
 
     _SUPPORTED_FILTERS = {
+        "eq": lambda column: column.__eq__,
         "gt": lambda column: column.__gt__,
         "lt": lambda column: column.__lt__,
         "gte": lambda column: column.__ge__,
@@ -471,6 +475,8 @@ class FastCRUD(
         "between": lambda column: column.between,
         "in": lambda column: column.in_,
         "not_in": lambda column: column.not_in,
+        "or": lambda column: column.or_,
+        "not": lambda column: column.not_,
     }
 
     def __init__(
@@ -491,51 +497,131 @@ class FastCRUD(
         self,
         operator: str,
         value: Any,
-    ) -> Optional[Callable[[str], Callable]]:
+    ) -> Optional[FilterCallable]:
         if operator in {"in", "not_in", "between"}:
             if not isinstance(value, (tuple, list, set)):
                 raise ValueError(f"<{operator}> filter must be tuple, list or set")
-        return self._SUPPORTED_FILTERS.get(operator)
+        return cast(Optional[FilterCallable], self._SUPPORTED_FILTERS.get(operator))
 
     def _parse_filters(
-        self, model: Optional[Union[type[ModelType], AliasedClass]] = None, **kwargs
+            self,
+            model: Optional[Union[type[ModelType], AliasedClass]] = None,
+            **kwargs
     ) -> list[ColumnElement]:
+        """Parse and convert filter arguments into SQLAlchemy filter conditions.
+
+        Args:
+            model: The model to apply filters to. Defaults to self.model
+            **kwargs: Filter arguments in the format field_name__operator=value
+
+        Returns:
+            List of SQLAlchemy filter conditions
+        """
         model = model or self.model
         filters = []
 
         for key, value in kwargs.items():
-            if "__" in key:
-                field_name, op = key.rsplit("__", 1)
-                column = getattr(model, field_name, None)
-                if column is None:
-                    raise ValueError(f"Invalid filter column: {field_name}")
-                if op == "or":
-                    or_filters = [
-                        sqlalchemy_filter(column)(or_value)
-                        for or_key, or_value in value.items()
-                        if (
-                            sqlalchemy_filter := self._get_sqlalchemy_filter(
-                                or_key, or_value
-                            )
-                        )
-                        is not None
-                    ]
-                    filters.append(or_(*or_filters))
-                else:
-                    sqlalchemy_filter = self._get_sqlalchemy_filter(op, value)
-                    if sqlalchemy_filter:
-                        filters.append(
-                            sqlalchemy_filter(column)(value)
-                            if op != "between"
-                            else sqlalchemy_filter(column)(*value)
-                        )
+            if "__" not in key:
+                filters.extend(self._handle_simple_filter(model, key, value))
+                continue
+
+            field_name, operator = key.rsplit("__", 1)
+            model_column = self._get_column(model, field_name)
+
+            if operator == "or":
+                filters.extend(self._handle_or_filter(model_column, value))
+            elif operator == "not":
+                filters.extend(self._handle_not_filter(model_column, value))
             else:
-                column = getattr(model, key, None)
-                if column is not None:
-                    filters.append(column == value)
+                filters.extend(self._handle_standard_filter(model_column, operator, value))
 
         return filters
 
+    def _handle_simple_filter(
+            self,
+            model: Union[type[ModelType], AliasedClass],
+            key: str,
+            value: Any
+    ) -> list[ColumnElement]:
+        """Handle simple equality filters (e.g., name='John')."""
+        col = getattr(model, key, None)
+        return [col == value] if col is not None else []
+
+    def _handle_or_filter(
+            self,
+            col: Column,
+            value: dict
+    ) -> list[ColumnElement]:
+        """Handle OR conditions (e.g., age__or={'gt': 18, 'lt': 65})."""
+        if not isinstance(value, dict):
+            raise ValueError("OR filter value must be a dictionary")
+
+        or_conditions = []
+        for or_op, or_value in value.items():
+            sqlalchemy_filter = self._get_sqlalchemy_filter(or_op, or_value)
+            if sqlalchemy_filter:
+                condition = (
+                    sqlalchemy_filter(col)(*or_value)
+                    if or_op == "between"
+                    else sqlalchemy_filter(col)(or_value)
+                )
+                or_conditions.append(condition)
+
+        return [or_(*or_conditions)] if or_conditions else []
+
+    def _handle_not_filter(
+            self,
+            col: Column,
+            value: dict
+    ) -> list[ColumnElement[bool]]:
+        """Handle NOT conditions (e.g., age__not={'eq': 20, 'between': (30, 40)})."""
+        if not isinstance(value, dict):
+            raise ValueError("NOT filter value must be a dictionary")
+
+        not_conditions = []
+        for not_op, not_value in value.items():
+            sqlalchemy_filter = self._get_sqlalchemy_filter(not_op, not_value)
+            if sqlalchemy_filter is None:
+                continue
+
+            condition = (
+                sqlalchemy_filter(col)(*not_value)
+                if not_op == "between"
+                else sqlalchemy_filter(col)(not_value)
+            )
+            not_conditions.append(condition)
+
+        return [and_(*(not_(cond) for cond in not_conditions))] if not_conditions else []
+
+    def _handle_standard_filter(
+            self,
+            col: Column[Any],
+            operator: str,
+            value: Any
+    ) -> list[ColumnElement[bool]]:
+        """Handle standard comparison operators (e.g., age__gt=18)."""
+        sqlalchemy_filter = self._get_sqlalchemy_filter(operator, value)
+        if sqlalchemy_filter is None:
+            return []
+
+        condition = (
+            sqlalchemy_filter(col)(*value)
+            if operator == "between"
+            else sqlalchemy_filter(col)(value)
+        )
+        return [condition]
+
+    def _get_column(
+            self,
+            model: Union[type[ModelType], AliasedClass],
+            field_name: str
+    ) -> Column[Any]:
+        """Get column from model, raising ValueError if not found."""
+        model_column = getattr(model, field_name, None)
+        if model_column is None:
+            raise ValueError(f"Invalid filter column: {field_name}")
+        return cast(Column[Any], model_column)
+
     def _apply_sorting(
         self,
         stmt: Select,
@@ -2080,11 +2166,9 @@ class FastCRUD(
                 db,
                 limit=10,
                 sort_column='registration_date',
-                sort_order='desc',
             )
 
             # Fetch the next set of records using the cursor from the first page
-            next_cursor = first_page['next_cursor']
             second_page = await user_crud.get_multi_by_cursor(
                 db,
                 cursor=next_cursor,
@@ -2102,14 +2186,12 @@ class FastCRUD(
                 limit=10,
                 sort_column='age',
                 sort_order='asc',
-                age__gt=30,
             )
             ```
 
             Fetch records excluding a specific username using cursor-based pagination:
 
             ```python
-            first_page = await user_crud.get_multi_by_cursor(
                 db,
                 limit=10,
                 sort_column='username',
@@ -2121,7 +2203,6 @@ class FastCRUD(
         Note:
             This method is designed for efficient pagination in large datasets and is ideal for infinite scrolling features.
             Make sure the column used for cursor pagination is indexed for performance.
-            This method assumes that your records can be ordered by a unique, sequential field (like `id` or `created_at`).
         """
         if limit == 0:
             return {"data": [], "next_cursor": None}


Test patch:
diff --git a/tests/sqlalchemy/crud/test_get_multi.py b/tests/sqlalchemy/crud/test_get_multi.py
index 3deb9f5..c877da3 100644
--- a/tests/sqlalchemy/crud/test_get_multi.py
+++ b/tests/sqlalchemy/crud/test_get_multi.py
@@ -272,3 +272,95 @@ async def test_read_items_with_advanced_filters(
     result = await crud.get_multi(async_session, name__startswith=name)
 
     assert len(result["data"]) == 0
+
+
+@pytest.mark.asyncio
+async def test_get_multi_or_filtering(async_session, test_model):
+    # Create specific test data for OR filtering
+    test_data = [
+        {"name": "Alice", "tier_id": 1, "category_id": 1},
+        {"name": "Bob", "tier_id": 2, "category_id": 1},
+        {"name": "Charlie", "tier_id": 3, "category_id": 1},
+        {"name": "David", "tier_id": 4, "category_id": 1},
+        {"name": "Alice2", "tier_id": 5, "category_id": 1},
+        {"name": "Frank", "tier_id": 6, "category_id": 1},
+    ]
+
+    for item in test_data:
+        async_session.add(test_model(**item))
+    await async_session.commit()
+
+    crud = FastCRUD(test_model)
+
+    # Test OR with simple conditions on tier_id
+    result = await crud.get_multi(
+        async_session,
+        tier_id__or={"eq": 1, "eq": 6}
+    )
+    assert len(result["data"]) > 0
+    assert all(item["tier_id"] in [1, 6] for item in result["data"])
+
+    # Test OR with range conditions on tier_id
+    result = await crud.get_multi(
+        async_session,
+        tier_id__or={"lt": 2, "gt": 5}
+    )
+    assert len(result["data"]) > 0
+    assert all(
+        item["tier_id"] < 2 or item["tier_id"] > 5
+        for item in result["data"]
+    )
+
+    # Test OR with like conditions on name
+    result = await crud.get_multi(
+        async_session,
+        name__or={"like": "Alice%", "like": "Frank%"}
+    )
+    assert len(result["data"]) > 0
+    assert all(
+        item["name"].startswith("Alice") or item["name"].startswith("Frank")
+        for item in result["data"]
+    )
+
+
+@pytest.mark.asyncio
+async def test_get_multi_not_filtering(async_session, test_model):
+    # Create specific test data for NOT filtering
+    test_data = [
+        {"name": "Alice", "tier_id": 1, "category_id": 1},
+        {"name": "Bob", "tier_id": 2, "category_id": 1},
+        {"name": "Charlie", "tier_id": 3, "category_id": 1},
+        {"name": "David", "tier_id": 4, "category_id": 1},
+        {"name": "Eve", "tier_id": 5, "category_id": 1},
+        {"name": "Frank", "tier_id": 6, "category_id": 1},
+    ]
+
+    for item in test_data:
+        async_session.add(test_model(**item))
+    await async_session.commit()
+
+    crud = FastCRUD(test_model)
+
+    # Test NOT with single condition
+    result = await crud.get_multi(
+        async_session,
+        name__not={"eq": "Alice"}
+    )
+    assert len(result["data"]) > 0
+    assert all(item["name"] != "Alice" for item in result["data"])
+
+    # Test NOT with multiple conditions
+    result = await crud.get_multi(
+        async_session,
+        tier_id__not={
+            "between": (1, 3),
+            "eq": 5
+        }
+    )
+    assert len(result["data"]) > 0
+    assert all(
+        not (1 <= item["tier_id"] <= 3) and item["tier_id"] != 5
+        for item in result["data"]
+    )
+    # Should only return records with tier_id = 4 or tier_id = 6
+    assert all(item["tier_id"] in [4, 6] for item in result["data"])