Hotfix/sqla1.4

src/pypnusershub/db/models.py

@@ -2,6 +2,8 @@
 
 from __future__ import unicode_literals, print_function, absolute_import, division
 
+from utils_flask_sqla.models import qfilter
+
 """
 mappings applications et utilisateurs
 """
@@ -101,7 +103,7 @@
                 Application,
                 Application.id_application == UserApplicationRight.id_application,
             )
-            .filter(Application.code_application == code_app)
+            .where(Application.code_application == code_app)
         )
 
 
@@ -125,7 +127,9 @@
     _password = db.Column("pass", db.Unicode)
     _password_plus = db.Column("pass_plus", db.Unicode)
     email = db.Column(db.Unicode)
-    id_organisme = db.Column(db.Integer, ForeignKey("utilisateurs.bib_organismes.id_organisme"))
+    id_organisme = db.Column(
+        db.Integer, ForeignKey("utilisateurs.bib_organismes.id_organisme")
+    )
     remarques = db.Column(db.Unicode)
     champs_addi = db.Column(JSONB)
     date_insert = db.Column(db.DateTime)
@@ -149,11 +153,13 @@
                 UserApplicationRight,
                 or_(
                     UserApplicationRight.id_role == self.id_role,
-                    UserApplicationRight.id_role.in_([role.id_role for role in self.groups]),
+                    UserApplicationRight.id_role.in_(
+                        [role.id_role for role in self.groups]
+                    ),
                 ),
             )
             .join(Profils, UserApplicationRight.id_profil == Profils.id_profil)
-            .filter(UserApplicationRight.id_application == get_current_app_id())
+            .where(UserApplicationRight.id_application == get_current_app_id())
         )
         return q.scalar() or 0
 
@@ -206,14 +212,37 @@
     def __str__(self):
         return self.identifiant or self.nom_complet
 
+    @qfilter
+    def filter_by_app(cls, code_app=None, **kwargs):
+        if code_app is None:
+            code_app = current_app.config["CODE_APPLICATION"]
+        query = kwargs["query"]
+        return (
+            query.outerjoin(cor_roles, User.id_role == cor_roles.c.id_role_utilisateur)
+            .outerjoin(
+                UserApplicationRight,
+                or_(
+                    UserApplicationRight.id_role == cor_roles.c.id_role_groupe,
+                    UserApplicationRight.id_role == User.id_role,
+                ),
+            )
+            .join(
+                Application,
+                Application.id_application == UserApplicationRight.id_application,
+            )
+            .where(Application.code_application == code_app)
+        ).whereclause
+
 
 @serializable
 class Organisme(db.Model):
     __tablename__ = "bib_organismes"
     __table_args__ = {"schema": "utilisateurs"}
 
     id_organisme = db.Column(db.Integer, primary_key=True)
-    uuid_organisme = db.Column(UUID(as_uuid=True), default=select(func.uuid_generate_v4()))
+    uuid_organisme = db.Column(
+        UUID(as_uuid=True), default=select(func.uuid_generate_v4())
+    )
     nom_organisme = db.Column(db.Unicode)
     adresse_organisme = db.Column(db.Unicode)
     cp_organisme = db.Column(db.Unicode)
@@ -223,7 +252,9 @@
     email_organisme = db.Column(db.Unicode)
     url_organisme = db.Column(db.Unicode)
     url_logo = db.Column(db.Unicode)
-    id_parent = db.Column(db.Integer, db.ForeignKey("utilisateurs.bib_organismes.id_organisme"))
+    id_parent = db.Column(
+        db.Integer, db.ForeignKey("utilisateurs.bib_organismes.id_organisme")
+    )
     additional_data = db.Column(JSONB, nullable=True, server_default="{}")
     members = db.relationship(User, backref="organisme")
 
@@ -261,7 +292,9 @@
     nom_profil = db.Column(db.Unicode)
     desc_profil = db.Column(db.Unicode)
 
-    applications = relationship("Application", secondary=profils_for_app, back_populates="profils")
+    applications = relationship(
+        "Application", secondary=profils_for_app, back_populates="profils"
+    )
 
 
 @serializable
@@ -278,7 +311,9 @@
     desc_application = db.Column(db.Unicode)
     id_parent = db.Column(db.Integer)
 
-    profils = relationship(Profils, secondary=profils_for_app, back_populates="applications")
+    profils = relationship(
+        Profils, secondary=profils_for_app, back_populates="applications"
+    )
 
     def __repr__(self):
         return "<Application {!r}>".format(self.nom_application)
@@ -288,7 +323,9 @@
 
     @staticmethod
     def get_application(nom_application):
-        return Application.query.filter(Application.nom_application == nom_application).one()
+        return db.session.execute(
+            select(Application).where(Application.nom_application == nom_application)
+        ).scalar_one()
 
 
 class ApplicationRight(db.Model):
@@ -316,12 +353,16 @@
 
     __tablename__ = "cor_role_app_profil"
     __table_args__ = {"schema": "utilisateurs"}  # , 'extend_existing': True}
-    id_role = db.Column(db.Integer, ForeignKey("utilisateurs.t_roles.id_role"), primary_key=True)
+    id_role = db.Column(
+        db.Integer, ForeignKey("utilisateurs.t_roles.id_role"), primary_key=True
+    )
     id_profil = db.Column(
         db.Integer, ForeignKey("utilisateurs.t_profils.id_profil"), primary_key=True
     )
     id_application = db.Column(
-        db.Integer, ForeignKey("utilisateurs.t_applications.id_application"), primary_key=True
+        db.Integer,
+        ForeignKey("utilisateurs.t_applications.id_application"),
+        primary_key=True,
     )
 
     role = relationship("User")
@@ -350,7 +391,9 @@
     nom_role = db.Column(db.Unicode)
     prenom_role = db.Column(db.Unicode)
     id_application = db.Column(
-        db.Integer, db.ForeignKey("utilisateurs.t_applications.id_application"), primary_key=True
+        db.Integer,
+        db.ForeignKey("utilisateurs.t_applications.id_application"),
+        primary_key=True,
     )
     id_organisme = db.Column(db.Integer)
     application = relationship("Application", backref="app_users")
@@ -387,7 +430,9 @@
     nom_role = db.Column(db.Unicode)
     prenom_role = db.Column(db.Unicode)
     id_application = db.Column(
-        db.Integer, db.ForeignKey("utilisateurs.t_applications.id_application"), primary_key=True
+        db.Integer,
+        db.ForeignKey("utilisateurs.t_applications.id_application"),
+        primary_key=True,
     )
     id_organisme = db.Column(db.Integer)
     identifiant = db.Column(db.Unicode)

src/pypnusershub/db/models_register.py

@@ -3,7 +3,7 @@
 
 from flask import current_app
 from pypnusershub.db.models import check_and_encrypt_password
-from sqlalchemy import or_
+from sqlalchemy import or_, select
 from sqlalchemy.dialects.postgresql import JSONB
 
 from .models import User, db as DB
@@ -25,13 +25,13 @@
     desc_role = DB.Column(DB.Unicode)
     password = DB.Column(DB.Unicode)
     pass_md5 = DB.Column(DB.Unicode)
-    email = DB.Column(DB.Unicode)    
+    email = DB.Column(DB.Unicode)
     id_organisme = DB.Column(DB.Integer)
     remarques = DB.Column(DB.Unicode)
     champs_addi = DB.Column(JSONB)
     date_insert = DB.Column(DB.DateTime)
     date_update = DB.Column(DB.DateTime)
-    
+
     def set_password(self, password, password_confirmation, md5):
         self.password, self.pass_md5 = check_and_encrypt_password(
             password, password_confirmation, md5
@@ -50,51 +50,55 @@
             is_valid = False
             msg += "E-mail is not valid. "
         # check if user or temp user exist with an email or login given
-        role = (
-            DB.session.query(User)
-            .filter(or_(User.email == self.email, User.identifiant == self.identifiant))
-            .first()
-        )
+        role = DB.session.scalars(
+            select(User).where(
+                or_(User.email == self.email, User.identifiant == self.identifiant)
+            )
+        ).first()
         if role:
             is_valid = False
             if role.email == self.email:
                 msg += (
-                    f"Un compte avec l'email {self.email} existe déjà. " +
-                    "S'il s'agit de votre email, vous pouvez faire une demande de renouvellement " +
-                    "de mot de passe via la page de login de GeoNature."
+                    f"Un compte avec l'email {self.email} existe déjà. "
+                    + "S'il s'agit de votre email, vous pouvez faire une demande de renouvellement "
+                    + "de mot de passe via la page de login de GeoNature."
                 )
             else:
                 msg += (
-                    f"Un compte avec l'identifiant {self.identifiant} existe déjà. " +
-                    "Veuillez choisir un identifiant différent."
+                    f"Un compte avec l'identifiant {self.identifiant} existe déjà. "
+                    + "Veuillez choisir un identifiant différent."
                 )
 
-        temp_role = (
-            DB.session.query(TempUser)
-            .filter(or_(TempUser.email == self.email, TempUser.identifiant == self.identifiant))
-            .first()
-        )
+        temp_role = DB.session.scalars(
+            select(TempUser)
+            .where(
+                or_(
+                    TempUser.email == self.email,
+                    TempUser.identifiant == self.identifiant,
+                )
+            )
+            .limit(1)
+        ).first()
         if temp_role:
             is_valid = False
             if temp_role.email == self.email:
                 msg += (
-                    f"Un compte en attente de validation avec l'email {self.email} existe déjà. "+
-                    "Merci de patienter le temps que votre demande soit traitée."
-
+                    f"Un compte en attente de validation avec l'email {self.email} existe déjà. "
+                    + "Merci de patienter le temps que votre demande soit traitée."
                 )
             else:
                 msg += (
-                    "Un compte en attente de validation avec l'identifiant " +
-                    f"{self.identifiant} existe déjà. " +
-                    "Veuillez choisir un identifiant différent."
+                    "Un compte en attente de validation avec l'identifiant "
+                    + f"{self.identifiant} existe déjà. "
+                    + "Veuillez choisir un identifiant différent."
                 )
 
         return (is_valid, msg)
 
     def as_dict(self, recursif=False, columns=(), depth=None):
         """
-            The signature of the function must be the as same the as_dict func 
-            from https://github.com/PnX-SI/Utils-Flask-SQLAlchemy
+        The signature of the function must be the as same the as_dict func
+        from https://github.com/PnX-SI/Utils-Flask-SQLAlchemy
         """
         return {
             "id_temp_user": self.id_temp_user,
@@ -117,7 +121,6 @@
 
 
 class CorRoleToken(DB.Model):
-
     __tablename__ = "cor_role_token"
     __table_args__ = {"schema": "utilisateurs", "extend_existing": True}
 
@@ -126,7 +129,7 @@
 
     def as_dict(self, recursif=False, columns=(), depth=None):
         """
-            The signature of the function must be the as same the as_dict func 
-            from https://github.com/PnX-SI/Utils-Flask-SQLAlchemy
+        The signature of the function must be the as same the as_dict func
+        from https://github.com/PnX-SI/Utils-Flask-SQLAlchemy
         """
         return {"id_role": self.id_role, "token": self.token}

src/pypnusershub/db/tools.py

@@ -18,6 +18,8 @@
 from pypnusershub.db import models
 from pypnusershub.utils import text_resource_stream, get_current_app_id
 
+from pypnusershub.env import db
+
 log = logging.getLogger(__name__)
 
 
@@ -71,28 +73,28 @@
 
 def load_fixtures(con_uri):
     with text_resource_stream("fixtures.sql", "pypnusershub.db") as sql_file:
-
         engine = sa.create_engine(con_uri)
         with engine.connect():
             for line in sql_file:
                 if line.strip():
                     engine.execute(line)
             engine.execute("COMMIT")
 
+
 def encode_token(payload):
-    expire = datetime.now() + timedelta(seconds=current_app.config['COOKIE_EXPIRATION'])
+    expire = datetime.now() + timedelta(seconds=current_app.config["COOKIE_EXPIRATION"])
     header = {
         "alg": "HS256",
         "exp": int(datetime.timestamp(expire)),
     }
     jwt = JsonWebToken(["HS256"])
-    key = current_app.config['SECRET_KEY'].encode("UTF-8")
+    key = current_app.config["SECRET_KEY"].encode("UTF-8")
     return jwt.encode(header, payload, key)
 
 
 def decode_token(payload):
     jwt = JsonWebToken(["HS256"])
-    key = current_app.config['SECRET_KEY'].encode("UTF-8")
+    key = current_app.config["SECRET_KEY"].encode("UTF-8")
     claims = jwt.decode(payload, key)
     claims.validate()
     return dict(claims)
@@ -121,11 +123,11 @@
             if id_app != id_app_from_config:
                 log.info("Invalid token: the token not corespoding to the current app")
                 raise UnreadableAccessRightsError("Token BadSignature", 403)
-        return (
-            models.AppUser.query.filter(models.AppUser.id_role == id_role)
-            .filter(models.AppUser.id_application == id_app)
-            .one()
-        )
+        return db.session.execute(
+            sa.select(models.AppUser)
+            .where(models.AppUser.id_role == id_role)
+            .where(models.AppUser.id_application == id_app)
+        ).scalar_one()
 
     except NoResultFound:
         raise UnreadableAccessRightsError(

src/pypnusershub/login_manager.py

@@ -7,12 +7,14 @@
 from pypnusershub.db.models import User
 from pypnusershub.db.tools import decode_token
 
+from pypnusershub.env import db
+
 login_manager = LoginManager()
 
 
 @login_manager.user_loader
 def load_user(user_id):
-    return User.query.get(user_id)
+    return db.session.get(User, user_id)
 
 
 @login_manager.request_loader
@@ -24,7 +26,7 @@
         return None
     try:
         user_dict = decode_token(jwt)
-        user = User.query.get(user_dict["id_role"])
+        user = db.session.get(User, user_dict["id_role"])
         g.login_via_request = True
         return user
     except (ExpiredTokenError, JoseError):