Repository: igorbenav/fastcrud (python)

Issue:
Add Automatic Filters for Auto Generated Endpoints
**Is your feature request related to a problem? Please describe.**
I would like to be able to filter get_multi endpoint.

**Describe the solution you'd like**
I want to be filtering and searching in the get_multi endpoint. This is such a common feature in most CRUD system that I consider relevant. This is also available in flask-muck.

**Describe alternatives you've considered**
Some out-of-the-box solution are out there:
https://github.com/arthurio/fastapi-filter (sqlalchemy + mongo)
https://github.com/OleksandrZhydyk/FastAPI-SQLAlchemy-Filters (specific sqlalchemy)

**Additional context**
Add any other context or screenshots about the feature request here.
Ifnot being implemented could an  example be provided?

Pull request:
[WIP] Filters in Automatic Endpoints
This is a work in progress, but it will close #15. This PR adds optional filters to automatic endpoints.

- [x] Create a class to validate passed filters
- [x] Validate that filters are indeed columns of the db
- [x] Create function to dynamically add filters to endpoints
- [x] Accept `filter_config` in both `EndpointCreator` and `crud_router`, dynamically add filters as query parameters
- [x] Decide whether `get_multi` and `get_paginated` only are changed by `filter_config` 
- [x] Get the type of the query parameters and pass it to FastAPI
- [x] Tests covering changes
- [x] Documentation updates

The usage idea after it is merged is something like:

## Passing Filters as a Dict to `crud_router`

```python
from fastapi import FastAPI
from fastcrud import crud_router, FilterConfig
from myapp.models import MyModel
from myapp.schemas import CreateMyModel, UpdateMyModel
from myapp.database import async_session

app = FastAPI()

router = crud_router(
    session=async_session,
    model=MyModel,
    create_schema=CreateMyModel,
    update_schema=UpdateMyModel,
    filter_config=FilterConfig(filters={"id": None, "name": "default"})
)
# Adds CRUD routes with filtering capabilities
app.include_router(router, prefix="/mymodel")
```

Explanation:
- The FilterConfig specifies that 'id' should be a query parameter with no default value and 'name' should be a query parameter with a default value of 'default'.
- When fetching multiple items, you can filter by these parameters.
- Example GET request: `/mymodel/get_multi?id=1&name=example`

## Passing Filters as Keyword Arguments to `EndpointCreator`

It's also possible to use FilterConfig with keyword arguments or with `EndpointCreator` instead of `crud_router`:
```python
from fastapi import FastAPI
from fastcrud import EndpointCreator, FilterConfig
from myapp.models import MyModel
from myapp.schemas import CreateMyModel, UpdateMyModel
from myapp.database import async_session

app = FastAPI()

# Using FilterConfig with keyword arguments
endpoint_creator_kw = EndpointCreator(
    session=async_session,
    model=MyModel,
    create_schema=CreateMyModel,
    update_schema=UpdateMyModel,
    filter_config=FilterConfig(id=None, name="default")
)
# Adds CRUD routes with filtering capabilities
endpoint_creator_kw.add_routes_to_router()
# Include the internal router into the FastAPI app
app.include_router(endpoint_creator_kw.router, prefix="/mymodel")
```

Of course one could also pass as keyword arguments to `crud_router` or as a dict to `EndpointCreator`.

Notes:
- for this first version, `get_multi` and `get_paginated` only are getting filters. If people request, we may rethink this
- for now, we're also accepting a `dict` being passed instead of a `FilterConfig`

Patch:
diff --git a/docs/advanced/endpoint.md b/docs/advanced/endpoint.md
index f6e64bf..d007402 100644
--- a/docs/advanced/endpoint.md
+++ b/docs/advanced/endpoint.md
@@ -438,6 +438,102 @@ This setup ensures that the soft delete functionality within your application ut
 
 By specifying custom column names for soft deletion, you can adapt FastCRUD to fit the design of your database models, providing a flexible solution for handling deleted records in a way that best suits your application's needs.
 
+## Using Filters in FastCRUD
+
+FastCRUD provides filtering capabilities, allowing you to filter query results based on various conditions. Filters can be applied to `read_multi` and `read_paginated` endpoints. This section explains how to configure and use filters in FastCRUD.
+
+### Defining Filters
+
+Filters are either defined using the `FilterConfig` class or just passed as a dictionary. This class allows you to specify default filter values and validate filter types. Here's an example of how to define filters for a model:
+
+```python
+from fastcrud import FilterConfig
+
+# Define filter configuration for a model
+filter_config = FilterConfig(
+    tier_id=None,  # Default filter value for tier_id
+    name=None  # Default filter value for name
+)
+```
+
+And the same thing using a `dict`:
+```python
+filter_config = {
+    "tier_id": None,  # Default filter value for tier_id
+    "name": None,  # Default filter value for name
+}
+```
+
+By using `FilterConfig` you get better error messages.
+
+### Applying Filters to Endpoints
+
+You can apply filters to your endpoints by passing the `filter_config` to the `crud_router` or `EndpointCreator`. Here's an example:
+
+```python
+from fastcrud import crud_router
+from yourapp.models import YourModel
+from yourapp.schemas import CreateYourModelSchema, UpdateYourModelSchema
+from yourapp.database import async_session
+
+# Apply filters using crud_router
+app.include_router(
+    crud_router(
+        session=async_session,
+        model=YourModel,
+        create_schema=CreateYourModelSchema,
+        update_schema=UpdateYourModelSchema,
+        filter_config=filter_config,  # Apply the filter configuration
+        path="/yourmodel",
+        tags=["YourModel"]
+    )
+)
+```
+
+### Using Filters in Requests
+
+Once filters are configured, you can use them in your API requests. Filters are passed as query parameters. Here's an example of how to use filters in a request to a paginated endpoint:
+
+```http
+GET /yourmodel/get_paginated?page=1&itemsPerPage=3&tier_id=1&name=Alice
+```
+
+### Custom Filter Validation
+
+The `FilterConfig` class includes a validator to check filter types. If an invalid filter type is provided, a `ValueError` is raised. You can customize the validation logic by extending the `FilterConfig` class:
+
+```python
+from fastcrud import FilterConfig
+from pydantic import ValidationError
+
+class CustomFilterConfig(FilterConfig):
+    @field_validator("filters")
+    def check_filter_types(cls, filters: dict[str, Any]) -> dict[str, Any]:
+        for key, value in filters.items():
+            if not isinstance(value, (type(None), str, int, float, bool)):
+                raise ValueError(f"Invalid default value for '{key}': {value}")
+        return filters
+
+try:
+    # Example of invalid filter configuration
+    invalid_filter_config = CustomFilterConfig(invalid_field=[])
+except ValidationError as e:
+    print(e)
+```
+
+### Handling Invalid Filter Columns
+
+FastCRUD ensures that filters are applied only to valid columns in your model. If an invalid filter column is specified, a `ValueError` is raised:
+
+```python
+try:
+    # Example of invalid filter column
+    invalid_filter_config = FilterConfig(non_existent_column=None)
+except ValueError as e:
+    print(e)  # Output: Invalid filter column 'non_existent_column': not found in model
+```
+
+
 ## Conclusion
 
 The `EndpointCreator` class in FastCRUD offers flexibility and control over CRUD operations and custom endpoint creation. By extending this class or using the `included_methods` and `deleted_methods` parameters, you can tailor your API's functionality to your specific requirements, ensuring a more customizable and streamlined experience.
diff --git a/docs/advanced/overview.md b/docs/advanced/overview.md
index d05f9f9..4c0228c 100644
--- a/docs/advanced/overview.md
+++ b/docs/advanced/overview.md
@@ -39,5 +39,10 @@ Explore different ways of joining models in FastCRUD with examples and tips.
 
 - [Joining Models](joins.md#applying-joins-in-fastcrud-methods)
 
+### 8. Filters in Automatic Endpoints
+Learn how to add query parameters to your `get_multi` and `get_paginated` endpoints.
+
+- [Filters in Endpoints](endpoint.md#defining-filters)
+
 ## Prerequisites
 Advanced usage assumes a solid understanding of the basic features and functionalities of our application. Knowledge of FastAPI, SQLAlchemy, and Pydantic is highly recommended to fully grasp the concepts discussed.
diff --git a/fastcrud/__init__.py b/fastcrud/__init__.py
index 2f3a130..e802ce8 100644
--- a/fastcrud/__init__.py
+++ b/fastcrud/__init__.py
@@ -5,6 +5,7 @@ from .crud.fast_crud import FastCRUD
 from .endpoint.endpoint_creator import EndpointCreator
 from .endpoint.crud_router import crud_router
 from .crud.helper import JoinConfig
+from .endpoint.helper import FilterConfig
 
 __all__ = [
     "FastCRUD",
@@ -13,4 +14,5 @@ __all__ = [
     "JoinConfig",
     "aliased",
     "AliasedClass",
+    "FilterConfig",
 ]
diff --git a/fastcrud/crud/fast_crud.py b/fastcrud/crud/fast_crud.py
index d4060c5..ab58dc6 100644
--- a/fastcrud/crud/fast_crud.py
+++ b/fastcrud/crud/fast_crud.py
@@ -187,6 +187,7 @@ class FastCRUD(
         updated_at_column: str = "updated_at",
     ) -> None:
         self.model = model
+        self.model_col_names = [col.key for col in model.__table__.columns]
         self.is_deleted_column = is_deleted_column
         self.deleted_at_column = deleted_at_column
         self.updated_at_column = updated_at_column
@@ -1690,7 +1691,7 @@ class FastCRUD(
                 f"Expected exactly one record to delete, found {total_count}."
             )
 
-        if self.is_deleted_column in self.model.__table__.columns:
+        if self.is_deleted_column in self.model_col_names:
             update_stmt = (
                 update(self.model)
                 .filter(*filters)
diff --git a/fastcrud/endpoint/crud_router.py b/fastcrud/endpoint/crud_router.py
index 59b58ec..7bc777f 100644
--- a/fastcrud/endpoint/crud_router.py
+++ b/fastcrud/endpoint/crud_router.py
@@ -7,6 +7,7 @@ from pydantic import BaseModel
 
 from .endpoint_creator import EndpointCreator
 from ..crud.fast_crud import FastCRUD
+from .helper import FilterConfig
 
 CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
 UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
@@ -37,6 +38,7 @@ def crud_router(
     deleted_at_column: str = "deleted_at",
     updated_at_column: str = "updated_at",
     endpoint_names: Optional[dict[str, str]] = None,
+    filter_config: Optional[Union[FilterConfig, dict]] = None,
 ) -> APIRouter:
     """
     Creates and configures a FastAPI router with CRUD endpoints for a given model.
@@ -70,6 +72,7 @@ def crud_router(
         endpoint_names: Optional dictionary to customize endpoint names for CRUD operations. Keys are operation types
                         ("create", "read", "update", "delete", "db_delete", "read_multi", "read_paginated"), and
                         values are the custom names to use. Unspecified operations will use default names.
+        filter_config: Optional FilterConfig instance or dictionary to configure filters for the `read_multi` and `read_paginated` endpoints.
 
     Returns:
         Configured APIRouter instance with the CRUD endpoints.
@@ -229,6 +232,60 @@ def crud_router(
             }
         )
         ```
+
+        Using FilterConfig with dict:
+        ```python
+        from fastapi import FastAPI
+        from fastcrud import crud_router
+        from myapp.models import MyModel
+        from myapp.schemas import CreateMyModel, UpdateMyModel
+        from myapp.database import async_session
+
+        app = FastAPI()
+
+        router = crud_router(
+            session=async_session,
+            model=MyModel,
+            create_schema=CreateMyModel,
+            update_schema=UpdateMyModel,
+            filter_config=FilterConfig(filters={"id": None, "name": "default"})
+        )
+        # Adds CRUD routes with filtering capabilities
+        app.include_router(router, prefix="/mymodel")
+
+        # Explanation:
+        # The FilterConfig specifies that 'id' should be a query parameter with no default value
+        # and 'name' should be a query parameter with a default value of 'default'.
+        # When fetching multiple items, you can filter by these parameters.
+        # Example GET request: /mymodel/get_multi?id=1&name=example
+        ```
+
+        Using FilterConfig with keyword arguments:
+        ```python
+        from fastapi import FastAPI
+        from fastcrud import crud_router
+        from myapp.models import MyModel
+        from myapp.schemas import CreateMyModel, UpdateMyModel
+        from myapp.database import async_session
+
+        app = FastAPI()
+
+        router = crud_router(
+            session=async_session,
+            model=MyModel,
+            create_schema=CreateMyModel,
+            update_schema=UpdateMyModel,
+            filter_config=FilterConfig(id=None, name="default")
+        )
+        # Adds CRUD routes with filtering capabilities
+        app.include_router(router, prefix="/mymodel")
+
+        # Explanation:
+        # The FilterConfig specifies that 'id' should be a query parameter with no default value
+        # and 'name' should be a query parameter with a default value of 'default'.
+        # When fetching multiple items, you can filter by these parameters.
+        # Example GET request: /mymodel/get_multi?id=1&name=example
+        ```
     """
     crud = crud or FastCRUD(
         model=model,
@@ -252,6 +309,7 @@ def crud_router(
         deleted_at_column=deleted_at_column,
         updated_at_column=updated_at_column,
         endpoint_names=endpoint_names,
+        filter_config=filter_config,
     )
 
     endpoint_creator_instance.add_routes_to_router(
diff --git a/fastcrud/endpoint/endpoint_creator.py b/fastcrud/endpoint/endpoint_creator.py
index 83c930d..f31b82d 100644
--- a/fastcrud/endpoint/endpoint_creator.py
+++ b/fastcrud/endpoint/endpoint_creator.py
@@ -1,5 +1,4 @@
-import inspect
-from typing import Dict, Type, TypeVar, Optional, Callable, Sequence, Union
+from typing import Type, TypeVar, Optional, Callable, Sequence, Union
 from enum import Enum
 
 from fastapi import Depends, Body, Query, APIRouter
@@ -13,10 +12,14 @@ from ..paginated.helper import compute_offset
 from ..paginated.response import paginated_response
 from .helper import (
     CRUDMethods,
+    FilterConfig,
     _extract_unique_columns,
     _get_primary_keys,
     _get_python_type,
     _inject_dependencies,
+    _apply_model_pk,
+    _create_dynamic_filters,
+    _get_column_types,
 )
 
 CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
@@ -25,35 +28,6 @@ UpdateSchemaInternalType = TypeVar("UpdateSchemaInternalType", bound=BaseModel)
 DeleteSchemaType = TypeVar("DeleteSchemaType", bound=BaseModel)
 
 
-def apply_model_pk(**pkeys: Dict[str, type]):
-    """
-    This decorator injects positional arguments into a fastCRUD endpoint.
-    It dynamically changes the endpoint signature and allows to use
-    multiple primary keys without defining them explicitly.
-    """
-
-    def wrapper(endpoint):
-        signature = inspect.signature(endpoint)
-        parameters = [
-            p
-            for p in signature.parameters.values()
-            if p.kind == inspect.Parameter.POSITIONAL_OR_KEYWORD
-        ]
-        extra_positional_params = [
-            inspect.Parameter(
-                name=k, annotation=v, kind=inspect.Parameter.POSITIONAL_ONLY
-            )
-            for k, v in pkeys.items()
-        ]
-
-        endpoint.__signature__ = signature.replace(
-            parameters=extra_positional_params + parameters
-        )
-        return endpoint
-
-    return wrapper
-
-
 class EndpointCreator:
     """
     A class to create and register CRUD endpoints for a FastAPI application.
@@ -79,6 +53,7 @@ class EndpointCreator:
         endpoint_names: Optional dictionary to customize endpoint names for CRUD operations. Keys are operation types
                         ("create", "read", "update", "delete", "db_delete", "read_multi", "read_paginated"), and
                         values are the custom names to use. Unspecified operations will use default names.
+        filter_config: Optional FilterConfig instance or dictionary to configure filters for the `read_multi` and `read_paginated` endpoints.
 
     Raises:
         ValueError: If both `included_methods` and `deleted_methods` are provided.
@@ -169,6 +144,64 @@ class EndpointCreator:
         )
         endpoint_creator.add_routes_to_router()
         ```
+
+        Using filter_config with dict:
+        ```python
+        from fastapi import FastAPI
+        from fastcrud import EndpointCreator, FilterConfig
+
+        from myapp.models import MyModel
+        from myapp.schemas import CreateMyModel, UpdateMyModel
+        from myapp.database import async_session
+
+        app = FastAPI()
+        endpoint_creator = EndpointCreator(
+            session=async_session,
+            model=MyModel,
+            create_schema=CreateMyModel,
+            update_schema=UpdateMyModel,
+            filter_config=FilterConfig(filters={"id": None, "name": "default"})
+        )
+        # Adds CRUD routes with filtering capabilities
+        endpoint_creator.add_routes_to_router()
+        # Include the internal router into the FastAPI app
+        app.include_router(endpoint_creator.router, prefix="/mymodel")
+
+        # Explanation:
+        # The FilterConfig specifies that 'id' should be a query parameter with no default value
+        # and 'name' should be a query parameter with a default value of 'default'.
+        # When fetching multiple items, you can filter by these parameters.
+        # Example GET request: /mymodel/get_multi?id=1&name=example
+        ```
+
+        Using filter_config with keyword arguments:
+        ```python
+        from fastapi import FastAPI
+        from fastcrud import EndpointCreator, FilterConfig
+
+        from myapp.models import MyModel
+        from myapp.schemas: CreateMyModel, UpdateMyModel
+        from myapp.database: async_session
+
+        app = FastAPI()
+        endpoint_creator = EndpointCreator(
+            session=async_session,
+            model=MyModel,
+            create_schema=CreateMyModel,
+            update_schema=UpdateMyModel,
+            filter_config=FilterConfig(id=None, name="default")
+        )
+        # Adds CRUD routes with filtering capabilities
+        endpoint_creator.add_routes_to_router()
+        # Include the internal router into the FastAPI app
+        app.include_router(endpoint_creator.router, prefix="/mymodel")
+
+        # Explanation:
+        # The FilterConfig specifies that 'id' should be a query parameter with no default value
+        # and 'name' should be a query parameter with a default value of 'default'.
+        # When fetching multiple items, you can filter by these parameters.
+        # Example GET request: /mymodel/get_multi?id=1&name=example
+        ```
     """
 
     def __init__(
@@ -186,6 +219,7 @@ class EndpointCreator:
         deleted_at_column: str = "deleted_at",
         updated_at_column: str = "updated_at",
         endpoint_names: Optional[dict[str, str]] = None,
+        filter_config: Optional[Union[FilterConfig, dict]] = None,
     ) -> None:
         self._primary_keys = _get_primary_keys(model)
         self._primary_keys_types = {
@@ -220,6 +254,20 @@ class EndpointCreator:
             "read_paginated": "get_paginated",
         }
         self.endpoint_names = {**self.default_endpoint_names, **(endpoint_names or {})}
+        if filter_config:
+            if isinstance(filter_config, dict):
+                filter_config = FilterConfig(**filter_config)
+            self._validate_filter_config(filter_config)
+        self.filter_config = filter_config
+        self.column_types = _get_column_types(model)
+
+    def _validate_filter_config(self, filter_config: FilterConfig) -> None:
+        model_columns = self.crud.model_col_names
+        for key in filter_config.filters.keys():
+            if key not in model_columns:
+                raise ValueError(
+                    f"Invalid filter column '{key}': not found in model '{self.model.__name__}' columns"
+                )
 
     def _create_item(self):
         """Creates an endpoint for creating items in the database."""
@@ -247,7 +295,7 @@ class EndpointCreator:
     def _read_item(self):
         """Creates an endpoint for reading a single item from the database."""
 
-        @apply_model_pk(**self._primary_keys_types)
+        @_apply_model_pk(**self._primary_keys_types)
         async def endpoint(db: AsyncSession = Depends(self.session), **pkeys):
             item = await self.crud.get(db, **pkeys)
             if not item:  # pragma: no cover
@@ -258,18 +306,21 @@ class EndpointCreator:
 
     def _read_items(self):
         """Creates an endpoint for reading multiple items from the database."""
+        dynamic_filters = _create_dynamic_filters(self.filter_config, self.column_types)
 
         async def endpoint(
             db: AsyncSession = Depends(self.session),
             offset: int = Query(0),
             limit: int = Query(100),
+            filters: dict = Depends(dynamic_filters),
         ):
-            return await self.crud.get_multi(db, offset=offset, limit=limit)
+            return await self.crud.get_multi(db, offset=offset, limit=limit, **filters)
 
         return endpoint
 
     def _read_paginated(self):
         """Creates an endpoint for reading multiple items from the database with pagination."""
+        dynamic_filters = _create_dynamic_filters(self.filter_config, self.column_types)
 
         async def endpoint(
             db: AsyncSession = Depends(self.session),
@@ -279,10 +330,11 @@ class EndpointCreator:
             items_per_page: int = Query(
                 10, alias="itemsPerPage", description="Number of items per page"
             ),
+            filters: dict = Depends(dynamic_filters),
         ):
             offset = compute_offset(page=page, items_per_page=items_per_page)
             crud_data = await self.crud.get_multi(
-                db, offset=offset, limit=items_per_page
+                db, offset=offset, limit=items_per_page, **filters
             )
 
             return paginated_response(
@@ -294,7 +346,7 @@ class EndpointCreator:
     def _update_item(self):
         """Creates an endpoint for updating an existing item in the database."""
 
-        @apply_model_pk(**self._primary_keys_types)
+        @_apply_model_pk(**self._primary_keys_types)
         async def endpoint(
             item: self.update_schema = Body(...),  # type: ignore
             db: AsyncSession = Depends(self.session),
@@ -307,7 +359,7 @@ class EndpointCreator:
     def _delete_item(self):
         """Creates an endpoint for deleting an item from the database."""
 
-        @apply_model_pk(**self._primary_keys_types)
+        @_apply_model_pk(**self._primary_keys_types)
         async def endpoint(db: AsyncSession = Depends(self.session), **pkeys):
             await self.crud.delete(db, **pkeys)
             return {"message": "Item deleted successfully"}  # pragma: no cover
@@ -323,7 +375,7 @@ class EndpointCreator:
         async session to permanently delete the item from the database.
         """
 
-        @apply_model_pk(**self._primary_keys_types)
+        @_apply_model_pk(**self._primary_keys_types)
         async def endpoint(db: AsyncSession = Depends(self.session), **pkeys):
             await self.crud.db_delete(db, **pkeys)
             return {
diff --git a/fastcrud/endpoint/helper.py b/fastcrud/endpoint/helper.py
index a61c873..a525c5b 100644
--- a/fastcrud/endpoint/helper.py
+++ b/fastcrud/endpoint/helper.py
@@ -1,14 +1,17 @@
-from typing import Optional, Union, Annotated, Sequence, Callable
+import inspect
+from typing import Optional, Union, Annotated, Sequence, Callable, TypeVar, Any
 import warnings
 
 from pydantic import BaseModel, Field
 from pydantic.functional_validators import field_validator
-from fastapi import Depends, params
+from fastapi import Depends, Query, params
 
-from sqlalchemy import Column, inspect
+from sqlalchemy import Column, inspect as sa_inspect
 from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.sql.elements import KeyedColumnElement
 
+F = TypeVar("F", bound=Callable[..., Any])
+
 
 class CRUDMethods(BaseModel):
     valid_methods: Annotated[
@@ -45,6 +48,28 @@ class CRUDMethods(BaseModel):
         return values
 
 
+class FilterConfig(BaseModel):
+    filters: Annotated[dict[str, Any], Field(default={})]
+
+    @field_validator("filters")
+    def check_filter_types(cls, filters: dict[str, Any]) -> dict[str, Any]:
+        for key, value in filters.items():
+            if not isinstance(value, (type(None), str, int, float, bool)):
+                raise ValueError(f"Invalid default value for '{key}': {value}")
+        return filters
+
+    def __init__(self, **kwargs: Any) -> None:
+        filters = kwargs.pop("filters", {})
+        filters.update(kwargs)
+        super().__init__(filters=filters)
+
+    def get_params(self) -> dict[str, Any]:
+        params = {}
+        for key, value in self.filters.items():
+            params[key] = Query(value)
+        return params
+
+
 def _get_primary_key(
     model: type[DeclarativeBase],
 ) -> Union[str, None]:  # pragma: no cover
@@ -54,7 +79,7 @@ def _get_primary_key(
 
 def _get_primary_keys(model: type[DeclarativeBase]) -> Sequence[Column]:
     """Get the primary key of a SQLAlchemy model."""
-    inspector = inspect(model).mapper
+    inspector = sa_inspect(model).mapper
     primary_key_columns: Sequence[Column] = inspector.primary_key
 
     return primary_key_columns
@@ -74,6 +99,15 @@ def _get_python_type(column: Column) -> Optional[type]:
             )
 
 
+def _get_column_types(model: type[DeclarativeBase]) -> dict[str, Union[type, None]]:
+    """Get a dictionary of column names and their corresponding Python types from a SQLAlchemy model."""
+    inspector = sa_inspect(model).mapper
+    column_types = {}
+    for column in inspector.columns:
+        column_types[column.name] = _get_python_type(column)
+    return column_types
+
+
 def _extract_unique_columns(
     model: type[DeclarativeBase],
 ) -> Sequence[KeyedColumnElement]:
@@ -84,7 +118,7 @@ def _extract_unique_columns(
 
 def _temporary_dependency_handling(
     funcs: Optional[Sequence[Callable]] = None,
-) -> Union[Sequence[params.Depends], None]: # pragma: no cover
+) -> Union[Sequence[params.Depends], None]:  # pragma: no cover
     """
     Checks if any function in the provided sequence is an instance of params.Depends.
     Issues a deprecation warning once if such instances are found, and returns the sequence if any params.Depends are found.
@@ -111,10 +145,70 @@ def _inject_dependencies(
 ) -> Optional[Sequence[params.Depends]]:
     """Wraps a list of functions in FastAPI's Depends."""
     temp_handling = _temporary_dependency_handling(funcs)
-    if temp_handling is not None: # pragma: no cover
+    if temp_handling is not None:  # pragma: no cover
         return temp_handling
 
     if funcs is not None:
         return [Depends(func) for func in funcs]
 
     return None
+
+
+def _apply_model_pk(**pkeys: dict[str, type]):
+    """
+    This decorator injects positional arguments into a fastCRUD endpoint.
+    It dynamically changes the endpoint signature and allows to use
+    multiple primary keys without defining them explicitly.
+    """
+
+    def wrapper(endpoint):
+        signature = inspect.signature(endpoint)
+        parameters = [
+            p
+            for p in signature.parameters.values()
+            if p.kind == inspect.Parameter.POSITIONAL_OR_KEYWORD
+        ]
+        extra_positional_params = [
+            inspect.Parameter(
+                name=k, annotation=v, kind=inspect.Parameter.POSITIONAL_ONLY
+            )
+            for k, v in pkeys.items()
+        ]
+
+        endpoint.__signature__ = signature.replace(
+            parameters=extra_positional_params + parameters
+        )
+        return endpoint
+
+    return wrapper
+
+
+def _create_dynamic_filters(
+    filter_config: Optional[FilterConfig], column_types: dict[str, type]
+) -> Callable[..., dict[str, Any]]:
+    if filter_config is None:
+        return lambda: {}
+
+    def filters(
+        **kwargs: Any,
+    ) -> dict[str, Any]:
+        filtered_params = {}
+        for key, value in kwargs.items():
+            if value is not None:
+                filtered_params[key] = value
+        return filtered_params
+
+    params = []
+    for key, value in filter_config.filters.items():
+        params.append(
+            inspect.Parameter(
+                key,
+                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                default=Query(value, alias=key),
+            )
+        )
+
+    sig = inspect.Signature(params)
+    setattr(filters, "__signature__", sig)
+
+    return filters


Test patch:
diff --git a/tests/sqlalchemy/conftest.py b/tests/sqlalchemy/conftest.py
index 18b325b..b245109 100644
--- a/tests/sqlalchemy/conftest.py
+++ b/tests/sqlalchemy/conftest.py
@@ -13,7 +13,7 @@ from sqlalchemy.sql import func
 
 from fastcrud.crud.fast_crud import FastCRUD
 from fastcrud.endpoint.crud_router import crud_router
-from fastcrud import EndpointCreator
+from fastcrud import EndpointCreator, FilterConfig
 
 
 class Base(DeclarativeBase):
@@ -22,7 +22,6 @@ class Base(DeclarativeBase):
 
 class MultiPkModel(Base):
     __tablename__ = "multi_pk"
-    # tests = relationship("ModelTest", back_populates="category")
     id = Column(Integer, primary_key=True)
     uuid = Column(String(32), primary_key=True)
     name = Column(String, unique=True)
@@ -369,6 +368,82 @@ def client(
     return TestClient(app)
 
 
+@pytest.fixture
+def filtered_client(
+    test_model,
+    create_schema,
+    update_schema,
+    delete_schema,
+):
+    app = FastAPI()
+
+    app.include_router(
+        crud_router(
+            session=get_session_local,
+            model=test_model,
+            crud=FastCRUD(test_model),
+            create_schema=create_schema,
+            update_schema=update_schema,
+            delete_schema=delete_schema,
+            filter_config=FilterConfig(tier_id=None, name=None),
+            path="/test",
+            tags=["test"],
+        )
+    )
+
+    return TestClient(app)
+
+
+@pytest.fixture
+def dict_filtered_client(
+    test_model,
+    create_schema,
+    update_schema,
+    delete_schema,
+):
+    app = FastAPI()
+
+    app.include_router(
+        crud_router(
+            session=get_session_local,
+            model=test_model,
+            crud=FastCRUD(test_model),
+            create_schema=create_schema,
+            update_schema=update_schema,
+            delete_schema=delete_schema,
+            filter_config={"tier_id": None, "name": None},
+            path="/test",
+            tags=["test"],
+        )
+    )
+
+    return TestClient(app)
+
+
+@pytest.fixture
+def invalid_filtered_client(
+    test_model,
+    create_schema,
+    update_schema,
+    delete_schema,
+):
+    filter_config = {"invalid_column": None}
+
+    with pytest.raises(
+        ValueError, match="Invalid filter column 'invalid_column': not found in model"
+    ):
+        EndpointCreator(
+            session=get_session_local,
+            model=test_model,
+            create_schema=create_schema,
+            update_schema=update_schema,
+            delete_schema=delete_schema,
+            filter_config=filter_config,
+            path="/test",
+            tags=["test"],
+        )
+
+
 @pytest.fixture
 def endpoint_creator(test_model) -> EndpointCreator:
     """Fixture to create an instance of EndpointCreator."""
diff --git a/tests/sqlalchemy/core/test_filter_config.py b/tests/sqlalchemy/core/test_filter_config.py
new file mode 100644
index 0000000..6fba41f
--- /dev/null
+++ b/tests/sqlalchemy/core/test_filter_config.py
@@ -0,0 +1,35 @@
+import pytest
+from fastcrud.endpoint.helper import FilterConfig
+
+
+def test_filter_config_with_invalid_default_value():
+    with pytest.raises(ValueError) as excinfo:
+        FilterConfig(filters={"valid_string": "value", "invalid_list": [1, 2, 3]})
+
+    assert "Invalid default value for 'invalid_list'" in str(excinfo.value)
+
+
+def test_filter_config_with_valid_default_values():
+    filter_config = FilterConfig(
+        filters={
+            "string": "value",
+            "int": 42,
+            "float": 3.14,
+            "bool": True,
+            "none": None,
+        }
+    )
+    assert filter_config.filters == {
+        "string": "value",
+        "int": 42,
+        "float": 3.14,
+        "bool": True,
+        "none": None,
+    }
+
+
+def test_filter_config_get_params():
+    filter_config = FilterConfig(filters={"string": "value", "int": 42})
+    params = filter_config.get_params()
+    assert params["string"].default == "value"
+    assert params["int"].default == 42
diff --git a/tests/sqlalchemy/endpoint/test_get_items.py b/tests/sqlalchemy/endpoint/test_get_items.py
index 020f4de..1f5ba11 100644
--- a/tests/sqlalchemy/endpoint/test_get_items.py
+++ b/tests/sqlalchemy/endpoint/test_get_items.py
@@ -19,3 +19,77 @@ async def test_read_items(client: TestClient, async_session, test_model, test_da
 
     test_item = test_data[0]
     assert any(item["name"] == test_item["name"] for item in data["data"])
+
+
+@pytest.mark.asyncio
+async def test_read_items_with_filters(
+    filtered_client: TestClient, async_session, test_model, test_data
+):
+    for data in test_data:
+        new_item = test_model(**data)
+        async_session.add(new_item)
+    await async_session.commit()
+
+    tier_id = 1
+    name = "Alice"
+
+    response = filtered_client.get(f"/test/get_multi?tier_id={tier_id}")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert len(data["data"]) > 0
+
+    for item in data["data"]:
+        assert item["tier_id"] == tier_id
+
+    response = filtered_client.get(f"/test/get_multi?name={name}")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert len(data["data"]) > 0
+
+    for item in data["data"]:
+        assert item["name"] == name
+
+
+@pytest.mark.asyncio
+async def test_read_items_with_dict_filter_config(
+    dict_filtered_client: TestClient, async_session, test_model, test_data
+):
+    for data in test_data:
+        new_item = test_model(**data)
+        async_session.add(new_item)
+    await async_session.commit()
+
+    tier_id = 1
+    response = dict_filtered_client.get(f"/test/get_multi?tier_id={tier_id}")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert len(data["data"]) > 0
+
+    for item in data["data"]:
+        assert item["tier_id"] == tier_id
+
+    name = "Alice"
+    response = dict_filtered_client.get(f"/test/get_multi?name={name}")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert len(data["data"]) > 0
+
+    for item in data["data"]:
+        assert item["name"] == name
+
+
+@pytest.mark.asyncio
+async def test_invalid_filter_column(invalid_filtered_client):
+    pass
diff --git a/tests/sqlalchemy/endpoint/test_get_paginated.py b/tests/sqlalchemy/endpoint/test_get_paginated.py
index 985b921..67ad59f 100644
--- a/tests/sqlalchemy/endpoint/test_get_paginated.py
+++ b/tests/sqlalchemy/endpoint/test_get_paginated.py
@@ -34,3 +34,56 @@ async def test_read_paginated(client: TestClient, async_session, test_model, tes
 
     assert data["page"] == page
     assert data["items_per_page"] == items_per_page
+
+
+@pytest.mark.asyncio
+async def test_read_paginated_with_filters(
+    filtered_client: TestClient, async_session, test_model, test_data
+):
+    for data in test_data:
+        new_item = test_model(**data)
+        async_session.add(new_item)
+    await async_session.commit()
+
+    page = 1
+    items_per_page = 5
+
+    tier_id = 1
+    response = filtered_client.get(
+        f"/test/get_paginated?page={page}&itemsPerPage={items_per_page}&tier_id={tier_id}"
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert "total_count" in data
+    assert "page" in data
+    assert "items_per_page" in data
+    assert "has_more" in data
+
+    assert len(data["data"]) > 0
+    assert len(data["data"]) <= items_per_page
+
+    for item in data["data"]:
+        assert item["tier_id"] == tier_id
+
+    name = "Alice"
+    response = filtered_client.get(
+        f"/test/get_paginated?page={page}&itemsPerPage={items_per_page}&name={name}"
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert "total_count" in data
+    assert "page" in data
+    assert "items_per_page" in data
+    assert "has_more" in data
+
+    assert len(data["data"]) > 0
+    assert len(data["data"]) <= items_per_page
+
+    for item in data["data"]:
+        assert item["name"] == name
diff --git a/tests/sqlmodel/conftest.py b/tests/sqlmodel/conftest.py
index 85ad173..b4f89cc 100644
--- a/tests/sqlmodel/conftest.py
+++ b/tests/sqlmodel/conftest.py
@@ -14,7 +14,7 @@ from sqlalchemy.sql import func
 
 from fastcrud.crud.fast_crud import FastCRUD
 from fastcrud.endpoint.crud_router import crud_router
-from fastcrud import EndpointCreator
+from fastcrud import EndpointCreator, FilterConfig
 
 
 class MultiPKModel(SQLModel, table=True):
@@ -318,7 +318,7 @@ def client(
 
     app.include_router(
         crud_router(
-            session=get_session_local,
+            session=lambda: local_session(),
             model=test_model,
             crud=FastCRUD(test_model),
             create_schema=create_schema,
@@ -331,7 +331,7 @@ def client(
 
     app.include_router(
         crud_router(
-            session=get_session_local,
+            session=lambda: local_session(),
             model=tier_model,
             crud=FastCRUD(tier_model),
             create_schema=tier_schema,
@@ -344,7 +344,7 @@ def client(
 
     app.include_router(
         crud_router(
-            session=get_session_local,
+            session=lambda: local_session(),
             model=multi_pk_model,
             crud=FastCRUD(multi_pk_model),
             create_schema=multi_pk_test_create_schema,
@@ -359,11 +359,87 @@ def client(
     return TestClient(app)
 
 
+@pytest.fixture
+def filtered_client(
+    test_model,
+    create_schema,
+    update_schema,
+    delete_schema,
+):
+    app = FastAPI()
+
+    app.include_router(
+        crud_router(
+            session=lambda: local_session(),
+            model=test_model,
+            crud=FastCRUD(test_model),
+            create_schema=create_schema,
+            update_schema=update_schema,
+            delete_schema=delete_schema,
+            filter_config=FilterConfig(tier_id=None, name=None),
+            path="/test",
+            tags=["test"],
+        )
+    )
+
+    return TestClient(app)
+
+
+@pytest.fixture
+def dict_filtered_client(
+    test_model,
+    create_schema,
+    update_schema,
+    delete_schema,
+):
+    app = FastAPI()
+
+    app.include_router(
+        crud_router(
+            session=lambda: local_session(),
+            model=test_model,
+            crud=FastCRUD(test_model),
+            create_schema=create_schema,
+            update_schema=update_schema,
+            delete_schema=delete_schema,
+            filter_config={"tier_id": None, "name": None},
+            path="/test",
+            tags=["test"],
+        )
+    )
+
+    return TestClient(app)
+
+
+@pytest.fixture
+def invalid_filtered_client(
+    test_model,
+    create_schema,
+    update_schema,
+    delete_schema,
+):
+    filter_config = {"invalid_column": None}
+
+    with pytest.raises(
+        ValueError, match="Invalid filter column 'invalid_column': not found in model"
+    ):
+        EndpointCreator(
+            session=lambda: local_session(),
+            model=test_model,
+            create_schema=create_schema,
+            update_schema=update_schema,
+            delete_schema=delete_schema,
+            filter_config=filter_config,
+            path="/test",
+            tags=["test"],
+        )
+
+
 @pytest.fixture
 def endpoint_creator(test_model) -> EndpointCreator:
     """Fixture to create an instance of EndpointCreator."""
     return EndpointCreator(
-        session=get_session_local,
+        session=lambda: local_session(),
         model=ModelTest,
         crud=FastCRUD(test_model),
         create_schema=CreateSchemaTest,
diff --git a/tests/sqlmodel/core/test_filter_config.py b/tests/sqlmodel/core/test_filter_config.py
new file mode 100644
index 0000000..6fba41f
--- /dev/null
+++ b/tests/sqlmodel/core/test_filter_config.py
@@ -0,0 +1,35 @@
+import pytest
+from fastcrud.endpoint.helper import FilterConfig
+
+
+def test_filter_config_with_invalid_default_value():
+    with pytest.raises(ValueError) as excinfo:
+        FilterConfig(filters={"valid_string": "value", "invalid_list": [1, 2, 3]})
+
+    assert "Invalid default value for 'invalid_list'" in str(excinfo.value)
+
+
+def test_filter_config_with_valid_default_values():
+    filter_config = FilterConfig(
+        filters={
+            "string": "value",
+            "int": 42,
+            "float": 3.14,
+            "bool": True,
+            "none": None,
+        }
+    )
+    assert filter_config.filters == {
+        "string": "value",
+        "int": 42,
+        "float": 3.14,
+        "bool": True,
+        "none": None,
+    }
+
+
+def test_filter_config_get_params():
+    filter_config = FilterConfig(filters={"string": "value", "int": 42})
+    params = filter_config.get_params()
+    assert params["string"].default == "value"
+    assert params["int"].default == 42
diff --git a/tests/sqlmodel/endpoint/test_get_items.py b/tests/sqlmodel/endpoint/test_get_items.py
index 020f4de..8184a9a 100644
--- a/tests/sqlmodel/endpoint/test_get_items.py
+++ b/tests/sqlmodel/endpoint/test_get_items.py
@@ -19,3 +19,76 @@ async def test_read_items(client: TestClient, async_session, test_model, test_da
 
     test_item = test_data[0]
     assert any(item["name"] == test_item["name"] for item in data["data"])
+
+
+@pytest.mark.asyncio
+async def test_read_items_with_filters(
+    filtered_client: TestClient, async_session, test_model, test_data
+):
+    for data in test_data:
+        new_item = test_model(**data)
+        async_session.add(new_item)
+    await async_session.commit()
+
+    tier_id = 1
+    response = filtered_client.get(f"/test/get_multi?tier_id={tier_id}")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert len(data["data"]) > 0
+
+    for item in data["data"]:
+        assert item["tier_id"] == tier_id
+
+    name = "Alice"
+    response = filtered_client.get(f"/test/get_multi?name={name}")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert len(data["data"]) > 0
+
+    for item in data["data"]:
+        assert item["name"] == name
+
+
+@pytest.mark.asyncio
+async def test_read_items_with_dict_filter_config(
+    dict_filtered_client: TestClient, async_session, test_model, test_data
+):
+    for data in test_data:
+        new_item = test_model(**data)
+        async_session.add(new_item)
+    await async_session.commit()
+
+    tier_id = 1
+    response = dict_filtered_client.get(f"/test/get_multi?tier_id={tier_id}")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert len(data["data"]) > 0
+
+    for item in data["data"]:
+        assert item["tier_id"] == tier_id
+
+    name = "Alice"
+    response = dict_filtered_client.get(f"/test/get_multi?name={name}")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert len(data["data"]) > 0
+
+    for item in data["data"]:
+        assert item["name"] == name
+
+
+@pytest.mark.asyncio
+async def test_invalid_filter_column(invalid_filtered_client):
+    pass
diff --git a/tests/sqlmodel/endpoint/test_get_paginated.py b/tests/sqlmodel/endpoint/test_get_paginated.py
index 985b921..67ad59f 100644
--- a/tests/sqlmodel/endpoint/test_get_paginated.py
+++ b/tests/sqlmodel/endpoint/test_get_paginated.py
@@ -34,3 +34,56 @@ async def test_read_paginated(client: TestClient, async_session, test_model, tes
 
     assert data["page"] == page
     assert data["items_per_page"] == items_per_page
+
+
+@pytest.mark.asyncio
+async def test_read_paginated_with_filters(
+    filtered_client: TestClient, async_session, test_model, test_data
+):
+    for data in test_data:
+        new_item = test_model(**data)
+        async_session.add(new_item)
+    await async_session.commit()
+
+    page = 1
+    items_per_page = 5
+
+    tier_id = 1
+    response = filtered_client.get(
+        f"/test/get_paginated?page={page}&itemsPerPage={items_per_page}&tier_id={tier_id}"
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert "total_count" in data
+    assert "page" in data
+    assert "items_per_page" in data
+    assert "has_more" in data
+
+    assert len(data["data"]) > 0
+    assert len(data["data"]) <= items_per_page
+
+    for item in data["data"]:
+        assert item["tier_id"] == tier_id
+
+    name = "Alice"
+    response = filtered_client.get(
+        f"/test/get_paginated?page={page}&itemsPerPage={items_per_page}&name={name}"
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+
+    assert "data" in data
+    assert "total_count" in data
+    assert "page" in data
+    assert "items_per_page" in data
+    assert "has_more" in data
+
+    assert len(data["data"]) > 0
+    assert len(data["data"]) <= items_per_page
+
+    for item in data["data"]:
+        assert item["name"] == name