refactor code base, update dependencies

Author: OleksandrZhydyk

fastapi_sa_orm_filter/__init__.py

@@ -2,4 +2,7 @@
 from fastapi_sa_orm_filter.main import FilterCore # noqa
 from fastapi_sa_orm_filter.operators import Operators as ops # noqa
 
-__version__ = "0.2.2"
+__version__ = "0.2.3"
+
+from .main import FilterCore as FilterCore # noqa
+from .operators import Operators as Operators # noqa

fastapi_sa_orm_filter/interfaces.py

@@ -0,0 +1,16 @@
+from abc import ABC, abstractmethod
+
+from fastapi_sa_orm_filter.dto import ParsedFilter
+from fastapi_sa_orm_filter.operators import Operators as ops
+
+
+class QueryParser(ABC):
+
+    @abstractmethod
+    def __init__(self, custom_filter: str, allowed_filters: dict[str, list[ops]]) -> None:
+        self.custom_filter = custom_filter
+        self.allowed_filters = allowed_filters
+
+    @abstractmethod
+    def get_parsed_filter(self) -> tuple[list[list[ParsedFilter]], list[str]] | tuple[list, list]:
+        pass

fastapi_sa_orm_filter/main.py

@@ -8,10 +8,12 @@
 from starlette import status
 from sqlalchemy.sql import Select
 
+from fastapi_sa_orm_filter.dto import ParsedFilter
 from fastapi_sa_orm_filter.exceptions import SAFilterOrmException
+from fastapi_sa_orm_filter.interfaces import QueryParser
 from fastapi_sa_orm_filter.operators import Operators as ops
-from fastapi_sa_orm_filter.parsers import FilterQueryParser, OrderByQueryParser
-from fastapi_sa_orm_filter.sa_expression_builder import SAFilterExpressionBuilder
+from fastapi_sa_orm_filter.parsers import StringQueryParser
+from fastapi_sa_orm_filter.sa_expression_builder import SAFilterExpressionBuilder, SAOrderByExpressionBuilder
 
 
 class FilterCore:
@@ -39,7 +41,7 @@ def __init__(
         """
         self.model = model
         self._allowed_filters = allowed_filters
-        self.select_query_part = select_query_part
+        self.select_sql_query = select_query_part
 
     def get_query(self, custom_filter: str) -> Select[Any]:
         """
@@ -63,56 +65,56 @@ def get_query(self, custom_filter: str) -> Select[Any]:
                         model.category == 'Medicine'
                     ).order_by(model.id.desc())
         """
-        split_query = self._split_by_order_by(custom_filter)
         try:
-            complete_query = self._get_complete_query(*split_query)
+            query_parser = self._get_query_parser(custom_filter)
+            filter_query, order_by_query = query_parser.get_parsed_filter()
+            complete_query = self._get_complete_query(filter_query, order_by_query)
         except SAFilterOrmException as e:
             raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=e.args[0])
         return complete_query
 
-    def _get_complete_query(self, filter_query_str: str, order_by_query_str: str | None = None) -> Select[Any]:
-        select_query_part = self.get_select_query_part()
-        filter_query_part = self._get_filter_query_part(filter_query_str)
-        complete_query = select_query_part.filter(*filter_query_part)
-        group_query_part = self.get_group_by_query_part()
-        if group_query_part:
-            complete_query = complete_query.group_by(*group_query_part)
-        if order_by_query_str is not None:
-            order_by_query = self.get_order_by_query_part(order_by_query_str)
-            complete_query = complete_query.order_by(*order_by_query)
+    def _get_complete_query(
+            self, filter_query: list[list[ParsedFilter]] | list, order_by_query: list[str] | list
+    ) -> Select[Any]:
+        select_sa_query = self.get_select_query_part()
+        filter_sa_query = self._get_filter_sa_query(filter_query)
+        group_by_sa_query = self._get_group_by_sa_query()
+        order_by_sa_query = self._get_order_by_sa_query(order_by_query)
+
+        complete_query = (
+            select_sa_query
+               .filter(*filter_sa_query)
+               .group_by(*group_by_sa_query)
+               .order_by(*order_by_sa_query)
+        )
         return complete_query
 
     def get_select_query_part(self) -> Select[Any]:
-        if self.select_query_part is not None:
-            return self.select_query_part
+        if self.select_sql_query is not None:
+            return self.select_sql_query
         return select(self.model)
 
-    def _get_filter_query_part(self, filter_query_str: str) -> list[Any]:
-        conditions = self._get_filter_query(filter_query_str)
-        if len(conditions) == 0:
-            return conditions
+    def _get_filter_sa_query(self, filter_query: list[list[ParsedFilter]] | list) -> list[BinaryExpression] | list:
+        if len(filter_query) == 0:
+            return []
+        sa_builder = SAFilterExpressionBuilder(self.model)
+        conditions = sa_builder.get_expressions(filter_query)
         return [or_(*conditions)]
 
-    def get_group_by_query_part(self) -> list:
-        return []
+    def _get_order_by_sa_query(self, order_by_query: list[str] | list) -> list[UnaryExpression]:
+        if len(order_by_query) == 0:
+            return []
+        order_by_parser = SAOrderByExpressionBuilder(self.model)
+        return order_by_parser.get_order_by_query(order_by_query)
 
-    def get_order_by_query_part(self, order_by_query_str: str) -> list[UnaryExpression]:
-        order_by_parser = OrderByQueryParser(self.model)
-        return order_by_parser.get_order_by_query(order_by_query_str)
+    def _get_group_by_sa_query(self) -> list[BinaryExpression] | list:
+        group_query_part = self.get_group_by_query_part()
+        if len(group_query_part) == 0:
+            return []
+        return group_query_part
 
-    def _get_filter_query(self, custom_filter: str) -> list[BinaryExpression]:
-        filter_conditions = []
-        if custom_filter == "":
-            return filter_conditions
+    def get_group_by_query_part(self) -> list:
+        return []
 
-        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_query_parser(self, custom_filter: str) -> QueryParser:
+        return StringQueryParser(custom_filter, self._allowed_filters)

fastapi_sa_orm_filter/parsers.py

@@ -1,69 +1,73 @@
-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.interfaces import QueryParser
 from fastapi_sa_orm_filter.operators import Operators as ops
-from fastapi_sa_orm_filter.operators import OrderSequence
 
 
-class OrderByQueryParser:
+class StringQueryParser(QueryParser):
+
+    def __init__(self, custom_filter: str, allowed_filters: dict[str, list[ops]]) -> None:
+        self.custom_filter = custom_filter
+        self.allowed_filters = allowed_filters
+
+    def get_parsed_filter(self) -> tuple[list[list[ParsedFilter]], list[str]] | tuple[list, list]:
+        parsed_filter = []
+        parsed_order_by = []
+
+        if self.custom_filter == "":
+            return parsed_filter, parsed_order_by
+
+        split_query = [query_part.strip("&") for query_part in self.custom_filter.split("order_by=")]
+
+        if len(split_query) > 2:
+            raise SAFilterOrmException("Use only one order_by directive")
+
+        parsed_filter = self._get_filter_query_part(split_query[0])
+
+        if len(split_query) == 2:
+            parsed_order_by = self._get_order_by_query_part(split_query[1])
+
+        return parsed_filter, parsed_order_by
+
+    def _get_filter_query_part(self, filter_query_str: str) -> list[list[ParsedFilter]] | list:
+        if filter_query_str == "":
+            return []
+        filter_parser = StringFilterQueryParser(self.allowed_filters)
+        return filter_parser.get_parsed_query(filter_query_str)
+
+    def _get_order_by_query_part(self, order_by_query_str: str) -> list[str] | list:
+        if order_by_query_str == "":
+            return []
+        order_by_parser = StringOrderByQueryParser()
+        return order_by_parser.get_order_by_query(order_by_query_str)
+
+
+class StringOrderByQueryParser:
     """
     Class parse order by part of request query string.
     """
-    def __init__(self, model: type[DeclarativeBase]) -> None:
-        self._model = model
-
-    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:
-            if '-' in field:
-                column = getattr(self._model, field.strip('-'))
-                order_by_query.append(getattr(column, OrderSequence.desc)())
-            else:
-                column = getattr(self._model, field.strip('+'))
-                order_by_query.append(getattr(column, OrderSequence.asc)())
-        return order_by_query
-
-    def _validate_order_by_fields(self, order_by_query_str: str) -> list[str]:
-        """
-        :return:
-            [
-                +field_name,
-                -field_name
-            ]
-        """
-        order_by_fields = order_by_query_str.split(",")
-        model_fields = self._model.__table__.columns.keys()
-        for field in order_by_fields:
-            field = field.strip('+').strip('-')
-            if field in model_fields:
-                continue
-            raise SAFilterOrmException(f"Incorrect order_by field name {field} for model {self._model.__name__}")
-        return order_by_fields
+    def get_order_by_query(self, order_by_query_str: str) -> list[str]:
+        return order_by_query_str.split(",")
 
 
-class FilterQueryParser:
+class StringFilterQueryParser:
     """
     Class parse filter part of request query string.
     """
 
     def __init__(
-            self, query: str,
-            allowed_filters: dict[str, list[ops]]
+            self, allowed_filters: dict[str, list[ops]]
     ) -> None:
-        self._query = query
         self._allowed_filters = allowed_filters
 
-    def get_parsed_query(self) -> list[list[ParsedFilter]]:
+    def get_parsed_query(self, filter_query_str: str) -> list[list[ParsedFilter]]:
         """
         :return:
             [
                 [ParsedFilter, ParsedFilter, ParsedFilter]
             ]
         """
-        and_blocks = self._parse_by_conjunctions()
+        and_blocks = self._parse_by_conjunctions(filter_query_str)
         parsed_query = []
         for and_block in and_blocks:
             parsed_and_blocks = []
@@ -74,7 +78,7 @@ def get_parsed_query(self) -> list[list[ParsedFilter]]:
             parsed_query.append(parsed_and_blocks)
         return parsed_query
 
-    def _parse_by_conjunctions(self) -> list[list[str]]:
+    def _parse_by_conjunctions(self, filter_query_str: str) -> list[list[str]]:
         """
         Split request query string by 'OR' and 'AND' conjunctions
         to divide query string to field's conditions
@@ -84,7 +88,7 @@ def _parse_by_conjunctions(self) -> list[list[str]]:
                     ['field_name__operator=value']
                 ]
         """
-        and_blocks = [block.split("&") for block in self._query.split("|")]
+        and_blocks = [block.split("&") for block in filter_query_str.split("|")]
         return and_blocks
 
     def _parse_expression(

fastapi_sa_orm_filter/sa_expression_builder.py

@@ -2,14 +2,14 @@
 from typing import Any
 
 import pydantic
-from pydantic import create_model
+from pydantic import create_model, BaseModel
 from pydantic._internal._model_construction import ModelMetaclass
-from sqlalchemy import inspect, BinaryExpression, and_
+from sqlalchemy import inspect, BinaryExpression, and_, UnaryExpression
 from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute
 from sqlalchemy_to_pydantic import sqlalchemy_to_pydantic
 
 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 Operators as ops, OrderSequence
 
 
 class SAFilterExpressionBuilder:
@@ -32,8 +32,8 @@ def get_expressions(self, parsed_filters) -> list[BinaryExpression]:
                     model = self.get_relation_model(and_filter.relation)
                     table = model.__tablename__
                 column = self.get_column(model, and_filter.field_name)
-                serialized_dict = self.serialize_expression_value(table, column, and_filter.operator, and_filter.value)
-                value = serialized_dict[column.name]
+                serialized_dict = self.serialize_expression_value(table, and_filter.field_name, and_filter.operator, and_filter.value)
+                value = serialized_dict[and_filter.field_name]
                 expr = self.get_orm_for_field(column, and_filter.operator, value)
                 and_expr.append(expr)
             or_expr.append(and_(*and_expr))
@@ -83,7 +83,7 @@ class model.__name__(BaseModel):
 
         return serializers
 
-    def get_relations_classes(self) -> list:
+    def get_relations_classes(self) -> list[type[DeclarativeBase]]:
         return [relation[1].mapper.class_ for relation in self._relationships]
 
     def get_orm_for_field(
@@ -97,7 +97,7 @@ def get_orm_for_field(
         return getattr(column, ops[operator].value)(value)
 
     def serialize_expression_value(
-            self, table: str, column: InstrumentedAttribute, operator: str, value: str
+            self, table: str, field_name: str, operator: str, value: str
     ) -> dict[str, Any]:
         """
         Serialize expression value from string to python type value,
@@ -112,14 +112,14 @@ def serialize_expression_value(
                 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)
+            return model_serializer(**{field_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}'")
 
     @staticmethod
-    def get_optional_pydantic_model(model, pydantic_serializer, is_list: bool = False):
+    def get_optional_pydantic_model(model, pydantic_serializer, is_list: bool = False) -> BaseModel:
         fields = {}
         for k, v in pydantic_serializer.model_fields.items():
             origin_annotation = getattr(v, 'annotation')
@@ -129,3 +129,37 @@ def get_optional_pydantic_model(model, pydantic_serializer, is_list: bool = Fals
                 fields[k] = (origin_annotation, None)
         pydantic_model = create_model(model.__name__, **fields)
         return pydantic_model
+
+
+class SAOrderByExpressionBuilder:
+
+    def __init__(self, model: type[DeclarativeBase]) -> None:
+        self._model = model
+
+    def get_order_by_query(self, order_by_query: list[str]) -> list[UnaryExpression]:
+        order_by_fields = self._validate_order_by_fields(order_by_query)
+        order_by_sql_query = []
+        for field in order_by_fields:
+            if '-' in field:
+                column = getattr(self._model, field.strip('-'))
+                order_by_sql_query.append(getattr(column, OrderSequence.desc)())
+            else:
+                column = getattr(self._model, field.strip('+'))
+                order_by_sql_query.append(getattr(column, OrderSequence.asc)())
+        return order_by_sql_query
+
+    def _validate_order_by_fields(self, order_by_fields: list[str]) -> list[str]:
+        """
+        :return:
+            [
+                +field_name,
+                -field_name
+            ]
+        """
+        model_fields = self._model.__table__.columns.keys()
+        for field in order_by_fields:
+            field = field.strip('+').strip('-')
+            if field in model_fields:
+                continue
+            raise SAFilterOrmException(f"Incorrect order_by field name {field} for model {self._model.__name__}")
+        return order_by_fields