Skip to content

Add multi-field OR filter functionality #218

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
May 10, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 43 additions & 1 deletion docs/advanced/filters.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@ results = await crud.get_multi(db, age__gt=18)

## OR Operations

### Single Field OR

Use the `__or` suffix to apply multiple conditions to the same field with OR logic:

```python
Expand All @@ -30,6 +32,24 @@ results = await crud.get_multi(
# Generates: WHERE age < 18 OR age > 65
```

### Multi-Field OR

Use the special `_or` parameter to apply OR conditions across multiple different fields:

```python
# Find users with name containing 'john' OR email containing 'john'
results = await crud.get_multi(
db,
_or={
"name__ilike": "%john%",
"email__ilike": "%john%"
}
)
# Generates: WHERE name ILIKE '%john%' OR email ILIKE '%john%'
```

This is particularly useful for implementing search functionality across multiple fields.

## NOT Operations

Use the `__not` suffix to negate multiple conditions on the same field:
Expand Down Expand Up @@ -69,14 +89,36 @@ results = await crud.get_multi(
}
)

# Text search with OR conditions
# Text search with OR conditions on a single field
results = await crud.get_multi(
db,
name__or={
"startswith": "A",
"endswith": "smith"
}
)

# Search across multiple fields with the same keyword
keyword = "john"
results = await crud.get_multi(
db,
_or={
"name__ilike": f"%{keyword}%",
"email__ilike": f"%{keyword}%",
"phone__ilike": f"%{keyword}%",
"address__ilike": f"%{keyword}%"
}
)

# Combining multi-field OR with regular filters
results = await crud.get_multi(
db,
is_active=True, # Regular filter applied to all results
_or={
"name__ilike": "%search term%",
"description__ilike": "%search term%"
}
)
```

## Error Handling
Expand Down
53 changes: 53 additions & 0 deletions fastcrud/crud/fast_crud.py
Original file line number Diff line number Diff line change
Expand Up @@ -520,6 +520,10 @@ def _parse_filters(
model = model or self.model
filters = []

# Check for special _or key for multi-field OR filtering
if "_or" in kwargs:
filters.extend(self._handle_multi_field_or_filter(model, kwargs.pop("_or")))

for key, value in kwargs.items():
if "__" not in key:
filters.extend(self._handle_simple_filter(model, key, value))
Expand Down Expand Up @@ -611,6 +615,55 @@ def _handle_standard_filter(
)
return [condition]

def _handle_multi_field_or_filter(
self,
model: Union[type[ModelType], AliasedClass],
value: dict
) -> list[ColumnElement]:
"""Handle OR conditions across multiple fields.

This method allows for OR conditions between different fields, such as:
_or={'name__ilike': '%keyword%', 'email__ilike': '%keyword%'}

Args:
model: The model to apply filters to
value: Dictionary of field conditions in the format {'field_name__operator': value}

Returns:
List containing a single SQLAlchemy OR condition combining all specified filters
"""
if not isinstance(value, dict): # pragma: no cover
raise ValueError("Multi-field OR filter value must be a dictionary")

or_conditions = []

for field_condition, condition_value in value.items():
# Handle simple equality conditions (no operator)
if "__" not in field_condition:
col = getattr(model, field_condition, None)
if col is not None:
or_conditions.append(col == condition_value)
continue

# Handle conditions with operators
field_name, operator = field_condition.rsplit("__", 1)
try:
model_column = self._get_column(model, field_name)

sqlalchemy_filter = self._get_sqlalchemy_filter(operator, condition_value)
if sqlalchemy_filter:
condition = (
sqlalchemy_filter(model_column)(*condition_value)
if operator == "between"
else sqlalchemy_filter(model_column)(condition_value)
)
or_conditions.append(condition)
except ValueError:
# Skip invalid columns
continue

return [or_(*or_conditions)] if or_conditions else []

def _get_column(
self,
model: Union[type[ModelType], AliasedClass],
Expand Down
138 changes: 138 additions & 0 deletions tests/sqlalchemy/crud/test_multi_field_or_filter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
import pytest

from fastcrud import FastCRUD


@pytest.mark.asyncio
async def test_multi_field_or_filter(async_session, test_model):
# Create specific test data for multi-field OR filtering
test_data = [
{"name": "Alice Johnson", "tier_id": 1, "category_id": 1},
{"name": "Bob Smith", "tier_id": 2, "category_id": 1},
{"name": "Charlie Brown", "tier_id": 3, "category_id": 2},
{"name": "David Jones", "tier_id": 4, "category_id": 2},
{"name": "Eve Williams", "tier_id": 5, "category_id": 1},
{"name": "Frank Miller", "tier_id": 6, "category_id": 2},
]

for item in test_data:
async_session.add(test_model(**item))
await async_session.commit()

crud = FastCRUD(test_model)

# Test multi-field OR with simple equality conditions
result = await crud.get_multi(
async_session,
_or={"name": "Alice Johnson", "tier_id": 2}
)
assert len(result["data"]) == 2
assert any(item["name"] == "Alice Johnson" for item in result["data"])
assert any(item["tier_id"] == 2 for item in result["data"])

# Test multi-field OR with operators
result = await crud.get_multi(
async_session,
_or={"name__startswith": "Alice", "tier_id__gt": 5}
)
assert len(result["data"]) > 0
for item in result["data"]:
assert item["name"].startswith("Alice") or item["tier_id"] > 5

# Test multi-field OR with LIKE operators
keyword = "li"
result = await crud.get_multi(
async_session,
_or={
"name__ilike": f"%{keyword}%",
"category_id__eq": 2
}
)
assert len(result["data"]) > 0
for item in result["data"]:
assert (keyword.lower() in item["name"].lower() or
item["category_id"] == 2)

# Test multi-field OR with mixed operators
result = await crud.get_multi(
async_session,
_or={
"tier_id__gt": 4,
"name__startswith": "Alice"
}
)
assert len(result["data"]) > 0
for item in result["data"]:
assert item["tier_id"] > 4 or item["name"].startswith("Alice")

# Test multi-field OR combined with regular filters
result = await crud.get_multi(
async_session,
category_id=1, # Regular filter applied to all results
_or={
"name__ilike": "%Alice%",
"tier_id__eq": 2
}
)
assert len(result["data"]) > 0
for item in result["data"]:
assert item["category_id"] == 1
assert ("alice" in item["name"].lower() or
item["tier_id"] == 2)

# Test with no matching results
result = await crud.get_multi(
async_session,
_or={
"name": "NonExistent",
"tier_id": 999
}
)
assert len(result["data"]) == 0


@pytest.mark.asyncio
async def test_multi_field_or_filter_client_example(async_session, test_model):
# Create test data to simulate a search across multiple fields
test_data = [
{"name": "Acme Corp", "tier_id": 1, "category_id": 1},
{"name": "XYZ Inc", "tier_id": 2, "category_id": 1},
{"name": "ABC Ltd", "tier_id": 3, "category_id": 2},
{"name": "Tech Solutions", "tier_id": 4, "category_id": 2},
]

for item in test_data:
async_session.add(test_model(**item))
await async_session.commit()

crud = FastCRUD(test_model)

# Test searching across multiple fields with a keyword
keyword = "corp"
result = await crud.get_multi(
async_session,
_or={
"name__ilike": f"%{keyword}%",
"tier_id__eq": 2
}
)

assert len(result["data"]) > 0
for item in result["data"]:
assert (keyword.lower() in item["name"].lower() or
item["tier_id"] == 2)

# Test with a different keyword
keyword = "tech"
result = await crud.get_multi(
async_session,
_or={
"name__ilike": f"%{keyword}%",
"category_id__eq": 1
}
)

assert len(result["data"]) > 0
for item in result["data"]:
assert (keyword.lower() in item["name"].lower() or
item["category_id"] == 1)