Merge branch 'hotfix/sqla1.4' of https://github.com/PnX-SI/Utils-Flas…

…k-SQLAlchemy into feat/sqlalchemy2.0

.github/workflows/pytest.yml

@@ -16,15 +16,9 @@ jobs:
 
     strategy:
       matrix:
-        python-version: [ '3.7', '3.9', '3.11' ]
-        sqlalchemy-version: [ '1.3', '1.4' ]
+        python-version: [ '3.9', '3.11' ]
+        sqlalchemy-version: [ '1.4' ]
         include:
-          - sqlalchemy-version: '1.3'
-            sqlalchemy-lt-version: '1.4'
-            flask-sqlalchemy-version: '2.0'
-            flask-sqlalchemy-lt-version: '3.0'
-            flask-version: '2.2'
-            flask-lt-version: '3.0'
           - sqlalchemy-version: '1.4'
             sqlalchemy-lt-version: '2.0'
             flask-sqlalchemy-version: '3.0'

docs/changelog.rst

@@ -2,12 +2,19 @@
 CHANGELOG
 =========
 
+0.3.7 (unreleased)
+------------------
+
+**🐛 Corrections**
+
+* 
+
 0.3.6 (2023-09-14)
 ------------------
 
 **🐛 Corrections**
 
-* Correction du `total_filtered` #40 by @lpofrec
+* Correction du ``total_filtered`` (#40 by @lpofrec)
 
 0.3.5 (2023-08-08)
 ------------------

src/utils_flask_sqla/models.py

@@ -1,15 +1,103 @@
-from .sqlalchemy import CustomSelect
-from flask_sqlalchemy.model import Model
+from sqlalchemy.sql.expression import BooleanClauseList, BinaryExpression
+from flask_sqlalchemy.model import DefaultMeta
+from sqlalchemy.sql import select, Select
 
+AUTHORIZED_WHERECLAUSE_TYPES = [bool, BooleanClauseList, BinaryExpression]
 
-class SelectModel(Model):
-    __abstract__ = True
 
-    @classmethod
-    @property
-    def select(cls):
-        if hasattr(cls, "__select_class__"):
-            select_cls = cls.__select_class__
-        else:
-            select_cls = CustomSelect
-        return select_cls._create_future_select(cls)  # SQLA 2.0: _create_future_select → _create
+def is_whereclause_compatible(object):
+    return any([isinstance(object, type_) for type_ in AUTHORIZED_WHERECLAUSE_TYPES])
+
+
+def qfilter(*args_dec, **kwargs_dec):
+    """
+    This decorator allows you to constrain a SQLAlchemy model method to return a whereclause (by default) or a query. If
+    its `query` is set to True and no query is given in a `query` parameter, it will create one with a simple select: `select(model)`. The latter
+    is accessible through `kwargs.get("query")` in the decorated method.
+
+    The decorated query requires the following minimum parameters (cls,**kwargs).
+
+    >>> from utils_flask_sqla.models import qfilter
+    >>> from sqlalchemy.sql import select
+    >>> class Station(NomenclaturesMixin, db.Model):
+            __tablename__ = "t_stations"
+            __table_args__ = {"schema": "pr_occhab"}
+            # If you wish the method to return a whereclause
+            @qfilter
+            def filter_by_params(cls,**kwargs):
+                filters = []
+                if "id_station" in kwargs:
+                    filters.append(Station.id_station == kwargs["id_station"])
+                return filters
+            # If you wish the method to return a query
+            @qfilter(query=True)
+            def filter_by_paramsQ(cls,**kwargs):
+                query = kwargs("query") #  select(Station)
+                if "id_station" in kwargs:
+                    query = query.filter_by(id_station=kwargs["id_station"])
+                return query
+
+    >>> query = Station.filter_by_paramsQ(id_station=1)
+    >>> query2 = select(Station).where(Station.filter_by_params(id_station=1))
+
+    Parameters
+    ----------
+    query : bool
+        decorated function must (or not) return a query (Select)
+
+    Returns
+    -------
+    function
+        decorated method
+
+    Raises
+    ------
+    ValueError
+        Method's class is not DefaultMeta class
+    ValueError
+        if query is True and return value of the decorated method is not Select
+    ValueError
+        if query is False and return value of the decorated method is not a : `bool` or sqlalchemy.sql.expression.BooleanClauseList` or `sqlalchemy.sql.expression.BinaryExpression`
+
+    """
+    if len(args_dec) == 1 and len(kwargs_dec) == 0 and callable(args_dec[0]):
+        return _qfilter()(args_dec[0])
+    else:
+        return _qfilter(*args_dec, **kwargs_dec)
+
+
+def _qfilter(query=False):
+    is_query = query
+
+    def _qfilter_decorator(method):
+        def _(*args, **kwargs):
+            # verify if class of the method is ORM model
+            sqla_class = args[0]
+            if not isinstance(sqla_class, DefaultMeta):
+                raise ValueError(
+                    "The decorated method's class must inherit from flask_sqlalchemy.model.DefaultMeta"
+                )
+
+            query = kwargs.get("query", None)
+
+            # if no query given
+            if query == None:
+                query = select(sqla_class)
+                kwargs["query"] = query
+            result = method(*args, **kwargs)
+
+            if is_query and not isinstance(result, Select):
+                raise ValueError("Your method must return a SQLAlchemy Select object ")
+
+            if not is_query and not is_whereclause_compatible(result):
+                raise ValueError(
+                    "Your method must return an object in the following types: {} ".format(
+                        ", ".join(map(lambda cls: cls.__name__, AUTHORIZED_WHERECLAUSE_TYPES))
+                    )
+                )
+            # if filter is wanted as where clause
+            return result
+
+        return classmethod(_)
+
+    return _qfilter_decorator

src/utils_flask_sqla/schema.py

@@ -1,14 +1,33 @@
 from marshmallow.fields import Nested
+from marshmallow_sqlalchemy.fields import RelatedList, Related
+
+# from flask_marshmallow.fields import RelatedList
 
 
 class SmartRelationshipsMixin:
     """
     This mixin automatically exclude from serialization:
-    - Nested fields
-    - all fields with exclude=True in their metadata (e.g. fields.String(metadata={'exclude': True}))
-    Adding Nested fields to only will serialize defaults fields and specified Nested fields.
-    Adding exclude=True fields to only will serialize only specified fields (default marshmallow behaviour).
-    You can use '+field_name' syntax to serialize excluded fields without excluding defaults fields.
+
+    * Nested, RelatedList and Related fields
+    * all fields with exclude=True in their metadata (e.g. ``fields.String(metadata={'exclude': True})``)
+    Adding only Nested fields to ``only`` will not exclude others fields and serialize specified Nested fields.
+    Adding exclude=True fields to ``only`` will serialize only specified fields (default marshmallow behaviour).
+    You can use '+field_name' syntax on `only` to serialize default excluded fields (with metadata exclude = True) without other fields.
+
+    Examples :
+
+    .. code-block:: python
+
+       class FooSchema(SmartRelationshipsMixin):
+           id = fields.Int()
+           name = field.Str()
+           default_excluded_field = fields.Str(metadata={"exclude": True})
+           relationship = fields.Nested(OtherSchema) # or field.RelatedList() / field.Related()
+
+       FooSchema().dump() -> {"id": 1, "name": "toto" }
+       FooSchema(only=["+default_excluded_field"]).dump() -> {"id": 1, "name": "toto", default_excluded_field: "test" }
+       FooSchema(only=["relationship"]).dump() -> {"id": 1, "name": "toto",  relationship : {OtherSchema...} }
+       FooSchema(only=["id", "relationship"]).dump() -> {"id": 1, relationship : {OtherSchema...} }
     """
 
     def __init__(self, *args, **kwargs):
@@ -19,7 +38,11 @@ def __init__(self, *args, **kwargs):
             # excluded fields at meta level are not even generated by auto-schema
             if field is None:
                 continue
-            if isinstance(field, Nested):
+            if (
+                isinstance(field, Nested)
+                or isinstance(field, RelatedList)
+                or isinstance(field, Related)
+            ):
                 nested_fields.add(name)
             elif field.metadata.get("exclude", False):
                 excluded_fields.add(name)
@@ -40,6 +63,7 @@ def __init__(self, *args, **kwargs):
         exclude = kwargs.pop("exclude", None)
         exclude = set(exclude) if exclude is not None else set()
         exclude |= (excluded_fields | nested_fields) - firstlevel_only
+
         # If only contains only nested & additional fields, we need to add included_fields to serialize nested, additional & included fields.
         # If only does not contains nested or additional fields, we do nothing and marshmallow will serialize only specified fields.
         if only and not firstlevel_only - nested_fields - additional_fields:

src/utils_flask_sqla/tests/test_qfilter.py

@@ -0,0 +1,81 @@
+import pytest
+from flask import Flask
+from sqlalchemy import func, and_
+
+from flask_sqlalchemy import SQLAlchemy
+
+from utils_flask_sqla.models import qfilter
+
+
+db = SQLAlchemy()
+
+
+class FooModel(db.Model):
+    pk = db.Column(db.Integer, primary_key=True)
+
+
+class BarModel(db.Model):
+    pk = db.Column(db.Integer, primary_key=True)
+
+    @qfilter
+    def where_pk(cls, pk, **kwargs):
+        return BarModel.pk == pk
+
+    @qfilter(query=True)
+    def where_pk_query(cls, pk, **kwargs):
+        query = kwargs["query"]
+        return query.where(BarModel.pk == pk)
+
+    @qfilter
+    def where_pk_list(cls, pk, **kwargs):
+        return and_(*[BarModel.pk == pk])
+
+
+@pytest.fixture(scope="session")
+def app():
+    app = Flask("utils-flask-sqla")
+    app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///"
+    db.init_app(app)
+    with app.app_context():
+        db.create_all()
+        yield app
+
+
+@pytest.fixture(scope="session")
+def foo(app):
+    foo = FooModel()
+    db.session.add(foo)
+    db.session.commit()
+    return foo
+
+
+@pytest.fixture(scope="session")
+def bar(app):
+    bar = BarModel()
+    db.session.add(bar)
+    db.session.commit()
+    return bar
+
+
+class TestQfilter:
+    def test_qfilter(self, bar):
+        assert db.session.scalars(BarModel.where_pk_query(bar.pk)).one_or_none() is bar
+        assert (
+            db.session.scalars(db.select(BarModel).where(BarModel.where_pk(bar.pk))).one_or_none()
+            is bar
+        )
+
+        assert db.session.scalars(BarModel.where_pk_query(bar.pk + 1)).one_or_none() is not bar
+        assert (
+            db.session.scalars(
+                db.select(BarModel).where(BarModel.where_pk(bar.pk + 1))
+            ).one_or_none()
+            is not bar
+        )
+
+        assert (
+            db.session.scalars(
+                db.select(BarModel).where(BarModel.where_pk_list(bar.pk))
+            ).one_or_none()
+            is bar
+        )