Add the join condition onclause support (#54)

* Add the join condition onclause support

* Update join_type field define

* Improve performance

* Update testcases

* Add testcases

* Update load strategy support

* Expose JoinConfig

* Add join config testcases

Author: wu-clan

pyproject.toml

@@ -59,6 +59,9 @@ order-by-type = true
 quote-style = "single"
 docstring-code-format = true
 
+[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

sqlalchemy_crud_plus/__init__.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 from .crud import CRUDPlus as CRUDPlus
+from .types import JoinConfig as JoinConfig
 
 __version__ = '1.10.0'

sqlalchemy_crud_plus/crud.py

@@ -181,7 +181,11 @@ async def count(
         if kwargs:
             filters.extend(parse_filters(self.model, **kwargs))
 
-        stmt = select(func.count()).select_from(self.model)
+        if isinstance(self.primary_key, list):
+            stmt = select(func.count()).select_from(self.model)
+        else:
+            stmt = select(func.count(self.primary_key)).select_from(self.model)
+
         if filters:
             stmt = stmt.where(*filters)
 
@@ -535,14 +539,14 @@ async def bulk_update_models(
     ) -> int:
         """
         Bulk update multiple instances with different data for each record.
-        Each update item should have 'pk' key and other fields to update.
 
         :param session: The SQLAlchemy async session
         :param objs: To save a list of Pydantic schemas or dict for data
         :param pk_mode: Primary key mode, when enabled, the data must contain the primary key data
         :param flush: If `True`, flush all object changes to the database
         :param commit: If `True`, commits the transaction immediately
-        :return: Total number of updated records
+        :param kwargs: Filter expressions using field__operator=value syntax
+        :return:
         """
         if not pk_mode:
             filters = parse_filters(self.model, **kwargs)

sqlalchemy_crud_plus/types.py

@@ -2,10 +2,11 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
-from typing import Literal, TypeVar
+from typing import Any, Literal, TypeVar
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict, Field
 from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm.util import AliasedClass
 from sqlalchemy.sql.base import ExecutableOption
 
 Model = TypeVar('Model', bound=DeclarativeBase)
@@ -51,7 +52,16 @@
     'full',
 ]
 
-JoinConditions = list[str] | dict[str, JoinType]
+
+class JoinConfig(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    model: type[Model] | AliasedClass
+    join_on: Any
+    join_type: JoinType = Field(default='inner')
+
+
+JoinConditions = list[str | JoinConfig] | dict[str, JoinType]
 
 LoadOptions = list[ExecutableOption]
 

sqlalchemy_crud_plus/utils.py

@@ -17,14 +17,13 @@
     load_only,
     noload,
     raiseload,
-    selectin_polymorphic,
     selectinload,
     subqueryload,
     undefer,
     undefer_group,
-    with_expression,
 )
 from sqlalchemy.orm.util import AliasedClass
+from sqlalchemy.sql.base import ExecutableOption
 from sqlalchemy.sql.operators import ColumnOperators
 from sqlalchemy.sql.schema import Column
 
@@ -35,7 +34,7 @@
     ModelColumnError,
     SelectOperatorError,
 )
-from sqlalchemy_crud_plus.types import JoinConditions, LoadStrategies, Model
+from sqlalchemy_crud_plus.types import JoinConditions, JoinConfig, LoadStrategies, Model
 
 _SUPPORTED_FILTERS = {
     # Comparison: https://docs.sqlalchemy.org/en/20/core/operators.html#comparison-operators
@@ -181,7 +180,7 @@ def _create_arithmetic_filters(column: Column, op: str, value: dict[str, Any]) -
     return arithmetic_filters
 
 
-def _create_and_filters(column: Column, op: str, value: Any) -> list[ColumnElement | None]:
+def _create_and_filters(column: Column, op: str, value: Any) -> list[ColumnElement[Any] | None]:
     """
     Create AND filter expressions.
 
@@ -197,7 +196,7 @@ def _create_and_filters(column: Column, op: str, value: Any) -> list[ColumnEleme
     return and_filters
 
 
-def parse_filters(model: type[Model] | AliasedClass, **kwargs) -> list[ColumnElement]:
+def parse_filters(model: type[Model] | AliasedClass, **kwargs) -> list[ColumnElement[Any]]:
     """
     Parse filter expressions from keyword arguments.
 
@@ -218,6 +217,9 @@ def parse_filters(model: type[Model] | AliasedClass, **kwargs) -> list[ColumnEle
         if field_name == '__or' and op == '':
             __or__filters = []
 
+            if not isinstance(value, dict):
+                raise SelectOperatorError('__or__ filter value must be a dictionary')
+
             for _key, _value in value.items():
                 if '__' not in _key:
                     _column = get_column(model, _key)
@@ -303,14 +305,16 @@ def apply_sorting(
     return stmt
 
 
-def build_load_strategies(model: type[Model], load_strategies: LoadStrategies | None) -> list:
+def build_load_strategies(model: type[Model], load_strategies: LoadStrategies | None) -> list[ExecutableOption]:
     """
     Build relationship loading strategy options.
 
     :param model: SQLAlchemy model class
     :param load_strategies: Loading strategies configuration
     :return:
     """
+    if load_strategies is None:
+        return []
 
     strategies_map = {
         'contains_eager': contains_eager,
@@ -325,10 +329,10 @@ def build_load_strategies(model: type[Model], load_strategies: LoadStrategies |
         # Load
         'defer': defer,
         'load_only': load_only,
-        'selectin_polymorphic': selectin_polymorphic,
+        # 'selectin_polymorphic': selectin_polymorphic,
         'undefer': undefer,
         'undefer_group': undefer_group,
-        'with_expression': with_expression,
+        # 'with_expression': with_expression,
     }
 
     options = []
@@ -359,7 +363,7 @@ def build_load_strategies(model: type[Model], load_strategies: LoadStrategies |
     return options
 
 
-def apply_join_conditions(model: type[Model], stmt: Select, join_conditions: JoinConditions | None):
+def apply_join_conditions(model: type[Model], stmt: Select, join_conditions: JoinConditions | None) -> Select:
     """
     Apply JOIN conditions to the query statement.
 
@@ -368,13 +372,24 @@ def apply_join_conditions(model: type[Model], stmt: Select, join_conditions: Joi
     :param join_conditions: JOIN conditions configuration
     :return:
     """
+    if join_conditions is None:
+        return stmt
+
     if isinstance(join_conditions, list):
-        for column in join_conditions:
-            try:
-                attr = getattr(model, column)
-                stmt = stmt.join(attr)
-            except AttributeError:
-                raise ModelColumnError(f'Invalid model column: {column}')
+        for v in join_conditions:
+            if isinstance(v, str):
+                try:
+                    attr = getattr(model, v)
+                    stmt = stmt.join(attr)
+                except AttributeError:
+                    raise ModelColumnError(f'Invalid model column: {v}')
+            elif isinstance(v, JoinConfig):
+                if v.join_type == 'inner':
+                    stmt = stmt.join(v.model, v.join_on)
+                elif v.join_type == 'left':
+                    stmt = stmt.join(v.model, v.join_on, isouter=True)
+                elif v.join_type == 'full':
+                    stmt = stmt.join(v.model, v.join_on, full=True)
 
     elif isinstance(join_conditions, dict):
         for column, join_type in join_conditions.items():
@@ -383,10 +398,10 @@ def apply_join_conditions(model: type[Model], stmt: Select, join_conditions: Joi
                 raise JoinConditionError(f'Invalid join type: {join_type}, only supports {allowed_join_types}')
             try:
                 attr = getattr(model, column)
-                if join_type == 'left':
-                    stmt = stmt.join(attr, isouter=True)
-                elif join_type == 'inner':
+                if join_type == 'inner':
                     stmt = stmt.join(attr)
+                elif join_type == 'left':
+                    stmt = stmt.join(attr, isouter=True)
                 elif join_type == 'full':
                     stmt = stmt.join(attr, full=True)
                 else:

tests/test_create.py

@@ -132,3 +132,32 @@ async def test_bulk_create_models_composite_keys(async_db_session: AsyncSession,
     assert results[0].id == 1000
     assert results[0].name == 'bulk_pks_1'
     assert results[0].sex == 'male'
+
+
+@pytest.mark.asyncio
+async def test_bulk_create_models_with_flush(async_db_session: AsyncSession, crud_ins: CRUDPlus[Ins]):
+    data = [
+        {'name': 'bulk_flush_1', 'del_flag': False, 'created_time': datetime.now()},
+        {'name': 'bulk_flush_2', 'del_flag': False, 'created_time': datetime.now()},
+    ]
+
+    async with async_db_session.begin():
+        results = await crud_ins.bulk_create_models(async_db_session, data, flush=True)
+
+    assert len(results) == 2
+    assert results[0].name == 'bulk_flush_1'
+    assert results[1].name == 'bulk_flush_2'
+
+
+@pytest.mark.asyncio
+async def test_bulk_create_models_with_commit(async_db_session: AsyncSession, crud_ins: CRUDPlus[Ins]):
+    data = [
+        {'name': 'bulk_commit_1', 'del_flag': False, 'created_time': datetime.now()},
+        {'name': 'bulk_commit_2', 'del_flag': False, 'created_time': datetime.now()},
+    ]
+
+    results = await crud_ins.bulk_create_models(async_db_session, data, commit=True)
+
+    assert len(results) == 2
+    assert results[0].name == 'bulk_commit_1'
+    assert results[1].name == 'bulk_commit_2'