refactor filter parser

Author: OleksandrZhydyk

.github/workflows/ci_filter.yml .github/workflows/CI.yml.github/workflows/ci_filter.yml renamed to .github/workflows/CI.yml

@@ -31,8 +31,7 @@ v 0.1.5
 ```shell
 from fastapi import FastAPI
 from fastapi.params import Query
-from fastapi_sa_orm_filter.main import FilterCore
-from fastapi_sa_orm_filter.operators import Operators as ops
+from fastapi_sa_orm_filter import FilterCore, ops
 
 from db.base import get_session
 from db.models import MyModel

README.md

@@ -1,3 +1,5 @@
 """FastAPI-SQLAlchemy filter, transform request query string to SQLAlchemy orm query"""
+from fastapi_sa_orm_filter.main import FilterCore # noqa
+from fastapi_sa_orm_filter.operators import Operators as ops # noqa
 
 __version__ = "0.2.2"

fastapi_sa_orm_filter/__init__.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class ParsedFilter:
+    field_name: str
+    operator: str
+    value: str
+    relation: str | None
+
+    @property
+    def has_relation(self) -> bool:
+        return bool(self.relation)

fastapi_sa_orm_filter/dto.py

@@ -1,22 +1,17 @@
-import json
-
 from typing import Any, Type
 
-import pydantic
 from fastapi import HTTPException
-from pydantic import create_model
-from pydantic._internal._model_construction import ModelMetaclass
-from sqlalchemy_to_pydantic import sqlalchemy_to_pydantic
-from sqlalchemy import select, inspect
-from sqlalchemy.orm import InstrumentedAttribute, DeclarativeBase
+from sqlalchemy import select
+from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.sql.elements import BinaryExpression, UnaryExpression
-from sqlalchemy.sql.expression import and_, or_
+from sqlalchemy.sql.expression import or_
 from starlette import status
 from sqlalchemy.sql import Select
 
 from fastapi_sa_orm_filter.exceptions import SAFilterOrmException
 from fastapi_sa_orm_filter.operators import Operators as ops
-from fastapi_sa_orm_filter.parsers import _FilterQueryParser, _OrderByQueryParser
+from fastapi_sa_orm_filter.parsers import FilterQueryParser, OrderByQueryParser
+from fastapi_sa_orm_filter.sa_expression_builder import SAFilterExpressionBuilder
 
 
 class FilterCore:
@@ -43,9 +38,7 @@ def __init__(
                 }
         """
         self.model = model
-        self.relationships = inspect(self.model).relationships.items()
         self._allowed_filters = allowed_filters
-        self._model_serializers = self._create_pydantic_serializers()
         self.select_query_part = select_query_part
 
     def get_query(self, custom_filter: str) -> Select[Any]:
@@ -104,108 +97,22 @@ def get_group_by_query_part(self) -> list:
         return []
 
     def get_order_by_query_part(self, order_by_query_str: str) -> list[UnaryExpression]:
-        order_by_parser = _OrderByQueryParser(self.model)
+        order_by_parser = OrderByQueryParser(self.model)
         return order_by_parser.get_order_by_query(order_by_query_str)
 
     def _get_filter_query(self, custom_filter: str) -> list[BinaryExpression]:
         filter_conditions = []
         if custom_filter == "":
             return filter_conditions
-        query_parser = _FilterQueryParser(custom_filter, self.model, self._allowed_filters)
-
-        for and_expressions in query_parser.get_parsed_query():
-            and_condition = []
-            for expression in and_expressions:
-                table, column, operator, value = expression
-                serialized_dict = self._format_expression(table, column, operator, value)
-                value = serialized_dict[column.name]
-                param = self._get_orm_for_field(column, operator, value)
-                and_condition.append(param)
-            filter_conditions.append(and_(*and_condition))
-        return filter_conditions
-
-    def _create_pydantic_serializers(self) -> dict[str, dict[str, ModelMetaclass]]:
-        """
-        Create two pydantic models (optional and list field types)
-        for value: str serialization
-
-        :return: {
-            'optional_model':
-                class model.__name__(BaseModel):
-                    field: Optional[type]
-            'list_model':
-                class model.__name__(BaseModel):
-                    field: Optional[List[type]]
-        }
-        """
-
-        models = [self.model]
-        models.extend(self._get_relations())
-
-        serializers = {}
-
-        for model in models:
-            pydantic_serializer = sqlalchemy_to_pydantic(model)
-            optional_model = self._get_optional_pydantic_model(model, pydantic_serializer)
-            optional_list_model = self._get_optional_pydantic_model(model, pydantic_serializer, is_list=True)
-
-            serializers[model.__tablename__] = {
-                "optional_model": optional_model, "optional_list_model": optional_list_model
-            }
-
-        return serializers
 
-    def _get_relations(self) -> list:
-        return [relation[1].mapper.class_ for relation in self.relationships]
-
-    def _get_orm_for_field(
-        self, column: InstrumentedAttribute, operator: str, value: Any
-    ) -> BinaryExpression:
-        """
-        Create SQLAlchemy orm expression for the field
-        """
-        if operator in [ops.between]:
-            param = getattr(column, ops[operator].value)(*value)
-        else:
-            param = getattr(column, ops[operator].value)(value)
-        return param
-
-    def _format_expression(
-        self, table: str, column: InstrumentedAttribute, operator: str, value: str
-    ) -> dict[str, Any]:
-        """
-        Serialize expression value from string to python type value,
-        according to db model types
-
-        :return: {'field_name': [value, value]}
-        """
-        value = value.split(",")
-        try:
-            if operator not in [ops.between, ops.in_]:
-                value = value[0]
-                model_serializer = self._model_serializers[table]["optional_model"]
-            else:
-                model_serializer = self._model_serializers[table]["optional_list_model"]
-            return model_serializer(**{column.name: value}).model_dump(exclude_none=True)
-        except pydantic.ValidationError as e:
-            raise SAFilterOrmException(json.loads(e.json()))
-        except ValueError:
-            raise SAFilterOrmException(f"Incorrect filter value '{value}'")
+        parser = FilterQueryParser(custom_filter, self._allowed_filters)
+        parsed_filters = parser.get_parsed_query()
+        sa_builder = SAFilterExpressionBuilder(self.model)
+        return sa_builder.get_expressions(parsed_filters)
 
     @staticmethod
     def _split_by_order_by(query) -> list:
         split_query = [query_part.strip("&") for query_part in query.split("order_by=")]
         if len(split_query) > 2:
             raise SAFilterOrmException("Use only one order_by directive")
         return split_query
-
-    def _get_optional_pydantic_model(self, model, pydantic_serializer, is_list: bool = False):
-        fields = {}
-        for k, v in pydantic_serializer.model_fields.items():
-            origin_annotation = getattr(v, 'annotation')
-            if is_list:
-                fields[k] = (list[origin_annotation], None)
-            else:
-                fields[k] = (origin_annotation, None)
-        pydantic_model = create_model(model.__name__, **fields)
-        return pydantic_model

fastapi_sa_orm_filter/main.py

@@ -1,23 +1,20 @@
-from typing import Optional, Tuple, Union, List, Any, Type
-
-from fastapi import HTTPException
-from sqlalchemy import inspect
-from sqlalchemy.orm import InstrumentedAttribute, DeclarativeBase
+from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.sql.elements import UnaryExpression
 
+from fastapi_sa_orm_filter.dto import ParsedFilter
 from fastapi_sa_orm_filter.exceptions import SAFilterOrmException
 from fastapi_sa_orm_filter.operators import Operators as ops
 from fastapi_sa_orm_filter.operators import Sequence
 
 
-class _OrderByQueryParser:
+class OrderByQueryParser:
     """
     Class parse order by part of request query string.
     """
-    def __init__(self, model: Type[DeclarativeBase]) -> None:
+    def __init__(self, model: type[DeclarativeBase]) -> None:
         self._model = model
 
-    def get_order_by_query(self, order_by_query_str: str) -> List[UnaryExpression]:
+    def get_order_by_query(self, order_by_query_str: str) -> list[UnaryExpression]:
         order_by_fields = self._validate_order_by_fields(order_by_query_str)
         order_by_query = []
         for field in order_by_fields:
@@ -47,33 +44,33 @@ def _validate_order_by_fields(self, order_by_query_str: str) -> list[str]:
         return order_by_fields
 
 
-class _FilterQueryParser:
+class FilterQueryParser:
     """
     Class parse filter part of request query string.
     """
 
-    def __init__(self, query: str, model: Type[DeclarativeBase], allowed_filters: dict[str, list[ops]]) -> None:
+    def __init__(
+            self, query: str,
+            allowed_filters: dict[str, list[ops]]
+    ) -> None:
         self._query = query
-        self._model = model
-        self._relationships = inspect(model).relationships.items()
         self._allowed_filters = allowed_filters
 
-    def get_parsed_query(self) -> list[list[Any]]:
+    def get_parsed_query(self) -> list[list[ParsedFilter]]:
         """
         :return:
             [
-                [[column, operator, value], [column, operator, value]],
-                [[column, operator, value]]
+                [ParsedFilter, ParsedFilter, ParsedFilter]
             ]
         """
         and_blocks = self._parse_by_conjunctions()
         parsed_query = []
         for and_block in and_blocks:
             parsed_and_blocks = []
             for expression in and_block:
-                table, column, operator, value = self._parse_expression(expression)
-                self._validate_query_params(column.name, operator)
-                parsed_and_blocks.append([table, column, operator, value])
+                parsed_filter = self._parse_expression(expression)
+                self._validate_query_params(parsed_filter.field_name, parsed_filter.operator)
+                parsed_and_blocks.append(parsed_filter)
             parsed_query.append(parsed_and_blocks)
         return parsed_query
 
@@ -92,13 +89,12 @@ def _parse_by_conjunctions(self) -> list[list[str]]:
 
     def _parse_expression(
         self, expression: str
-    ) -> Union[Tuple[str, InstrumentedAttribute, str, str], HTTPException]:
-        model = self._model
-        table = self._model.__tablename__
+    ) -> ParsedFilter:
+        relation = None
         try:
             field_name, condition = expression.split("__")
             if "." in field_name:
-                model, table, field_name = self._get_relation_model(field_name)
+                relation, field_name = self._get_relation_model(field_name)
             operator, value = condition.split("=")
         except ValueError:
             raise SAFilterOrmException(
@@ -108,23 +104,14 @@ def _parse_expression(
                 "or '{relation}.{field_name}__{condition}={value}{conjunction}'",
             )
 
-        column = getattr(model, field_name, None)
-
-        if not column:
-            raise SAFilterOrmException(f"DB model {model.__name__} doesn't have field '{field_name}'")
-        return table, column, operator, value
+        return ParsedFilter(field_name=field_name, operator=operator, value=value, relation=relation)
 
-    def _get_relation_model(self, field_name: str) -> tuple[DeclarativeBase, str, str]:
-        relation, field_name = field_name.split(".")
-        for relationship in self._relationships:
-            if relationship[0] == relation:
-                model = relationship[1].mapper.class_
-                return model, model.__tablename__, field_name
-        raise SAFilterOrmException(f"Can not find relation {relation} in {self._model.__name__} model")
+    def _get_relation_model(self, field_name: str) -> list[str]:
+        return field_name.split(".")
 
     def _validate_query_params(
         self, field_name: str, operator: str
-    ) -> Optional[HTTPException]:
+    ) -> None:
         """
         Check expression on valid and allowed field_name and operator
         """