From 87889142e0e8d9f262ac3dec6852b9d500d10c0a Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Mon, 24 Apr 2023 22:52:17 +0200
Subject: [PATCH 1/8] Fixed compatibility with SQLAlchemy >= 2.0 in the `db`
 plugin.

---
 platypush/plugins/db/__init__.py | 44 +++++++++++++++++++++++---------
 setup.py                         |  2 +-
 2 files changed, 33 insertions(+), 13 deletions(-)

diff --git a/platypush/plugins/db/__init__.py b/platypush/plugins/db/__init__.py
index e777ca169d..e92c2fe7e6 100644
--- a/platypush/plugins/db/__init__.py
+++ b/platypush/plugins/db/__init__.py
@@ -3,7 +3,12 @@ from contextlib import contextmanager
 from multiprocessing import RLock
 from typing import Optional, Generator, Union
 
-from sqlalchemy import create_engine, Table, MetaData
+from sqlalchemy import (
+    create_engine,
+    Table,
+    MetaData,
+    __version__ as sa_version,
+)
 from sqlalchemy.engine import Engine
 from sqlalchemy.exc import CompileError
 from sqlalchemy.orm import Session, sessionmaker, scoped_session
@@ -14,6 +19,21 @@ from platypush.plugins import Plugin, action
 session_locks = {}
 
 
+@contextmanager
+def conn_begin(conn):
+    """
+    Utility method to deal with `autobegin` being enabled on SQLAlchemy 2.0 but
+    not on earlier version.
+    """
+    sa_maj_ver = int(sa_version.split('.')[0])
+    if sa_maj_ver < 2:
+        yield conn.begin()
+    else:
+        yield conn._transaction
+
+    conn.commit()
+
+
 class DbPlugin(Plugin):
     """
     Database plugin. It allows you to programmatically select, insert, update
@@ -101,7 +121,7 @@ class DbPlugin(Plugin):
         engine = self.get_engine(engine, *args, **kwargs)
 
         with engine.connect() as connection:
-            connection.execute(statement)
+            connection.execute(text(statement))
 
     def _get_table(self, table, engine=None, *args, **kwargs):
         if not engine:
@@ -115,7 +135,7 @@ class DbPlugin(Plugin):
             try:
                 n_tries += 1
                 metadata = MetaData()
-                table = Table(table, metadata, autoload=True, autoload_with=engine)
+                table = Table(table, metadata, autoload_with=engine)
                 db_ok = True
             except Exception as e:
                 last_error = e
@@ -139,7 +159,7 @@ class DbPlugin(Plugin):
         engine=None,
         data: Optional[dict] = None,
         *args,
-        **kwargs
+        **kwargs,
     ):
         """
         Returns rows (as a list of hashes) given a query.
@@ -219,7 +239,7 @@ class DbPlugin(Plugin):
             query = table.select()
 
             if filter:
-                for (k, v) in filter.items():
+                for k, v in filter.items():
                     query = query.where(self._build_condition(table, k, v))
 
         if query is None:
@@ -246,7 +266,7 @@ class DbPlugin(Plugin):
         key_columns=None,
         on_duplicate_update=False,
         *args,
-        **kwargs
+        **kwargs,
     ):
         """
         Inserts records (as a list of hashes) into a table.
@@ -324,7 +344,7 @@ class DbPlugin(Plugin):
                     connection, table, records, key_columns
                 )
 
-            with connection.begin():
+            with conn_begin(connection):
                 if insert_records:
                     insert = table.insert().values(insert_records)
                     ret = self._execute_try_returning(connection, insert)
@@ -394,7 +414,7 @@ class DbPlugin(Plugin):
             values = {k: v for (k, v) in record.items() if k not in key_columns}
             update = table.update()
 
-            for (k, v) in key.items():
+            for k, v in key.items():
                 update = update.where(self._build_condition(table, k, v))
 
             update = update.values(**values)
@@ -498,18 +518,18 @@ class DbPlugin(Plugin):
 
         engine = self.get_engine(engine, *args, **kwargs)
 
-        with engine.connect() as connection, connection.begin():
+        with engine.connect() as connection, conn_begin(connection):
             for record in records:
                 table, engine = self._get_table(table, engine=engine, *args, **kwargs)
                 delete = table.delete()
 
-                for (k, v) in record.items():
+                for k, v in record.items():
                     delete = delete.where(self._build_condition(table, k, v))
 
                 connection.execute(delete)
 
     def create_all(self, engine, base):
-        with (self.get_session(engine, locked=True) as session, session.begin()):
+        with self.get_session(engine, locked=True) as session, session.begin():
             base.metadata.create_all(session.connection())
 
     @contextmanager
@@ -523,7 +543,7 @@ class DbPlugin(Plugin):
             # Mock lock
             lock = RLock()
 
-        with (lock, engine.connect() as conn, conn.begin()):
+        with lock, engine.connect() as conn, conn_begin(conn):
             session = scoped_session(
                 sessionmaker(
                     expire_on_commit=False,
diff --git a/setup.py b/setup.py
index eb77d9ec85..48e64f1b84 100755
--- a/setup.py
+++ b/setup.py
@@ -57,7 +57,7 @@ setup(
         'redis',
         'requests',
         'croniter',
-        'sqlalchemy<2.0.0',
+        'sqlalchemy',
         'websockets',
         'websocket-client',
         'wheel',

From 91df18f7b55fd9d20ca395b074eb6af713d0789a Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Mon, 24 Apr 2023 23:21:39 +0200
Subject: [PATCH 2/8] Better way to import `declarative_base` from SQLAlchemy.

Import `declarative_base` in a way that is compatible with any
SQLAlchemy version between 1.3 and 2.x.
---
 platypush/backend/covid19/__init__.py          | 3 ++-
 platypush/backend/github/__init__.py           | 3 ++-
 platypush/backend/http/request/rss/__init__.py | 3 ++-
 platypush/backend/mail/__init__.py             | 4 +++-
 platypush/common/db.py                         | 9 ++++++++-
 platypush/plugins/media/search/local.py        | 3 ++-
 6 files changed, 19 insertions(+), 6 deletions(-)

diff --git a/platypush/backend/covid19/__init__.py b/platypush/backend/covid19/__init__.py
index 1be1db0165..5f7444970f 100644
--- a/platypush/backend/covid19/__init__.py
+++ b/platypush/backend/covid19/__init__.py
@@ -3,9 +3,10 @@ import os
 from typing import Optional, Union, List, Dict, Any
 
 from sqlalchemy import create_engine, Column, Integer, String, DateTime
-from sqlalchemy.orm import sessionmaker, scoped_session, declarative_base
+from sqlalchemy.orm import sessionmaker, scoped_session
 
 from platypush.backend import Backend
+from platypush.common.db import declarative_base
 from platypush.config import Config
 from platypush.context import get_plugin
 from platypush.message.event.covid19 import Covid19UpdateEvent
diff --git a/platypush/backend/github/__init__.py b/platypush/backend/github/__init__.py
index 0a1bc3e675..6922db3915 100644
--- a/platypush/backend/github/__init__.py
+++ b/platypush/backend/github/__init__.py
@@ -6,9 +6,10 @@ from typing import Optional, List
 
 import requests
 from sqlalchemy import create_engine, Column, String, DateTime
-from sqlalchemy.orm import sessionmaker, scoped_session, declarative_base
+from sqlalchemy.orm import sessionmaker, scoped_session
 
 from platypush.backend import Backend
+from platypush.common.db import declarative_base
 from platypush.config import Config
 from platypush.message.event.github import (
     GithubPushEvent,
diff --git a/platypush/backend/http/request/rss/__init__.py b/platypush/backend/http/request/rss/__init__.py
index 7ca6d9c64b..6c624ae493 100644
--- a/platypush/backend/http/request/rss/__init__.py
+++ b/platypush/backend/http/request/rss/__init__.py
@@ -12,10 +12,11 @@ from sqlalchemy import (
     ForeignKey,
 )
 
-from sqlalchemy.orm import sessionmaker, scoped_session, declarative_base
+from sqlalchemy.orm import sessionmaker, scoped_session
 from sqlalchemy.sql.expression import func
 
 from platypush.backend.http.request import HttpRequest
+from platypush.common.db import declarative_base
 from platypush.config import Config
 from platypush.context import get_plugin
 from platypush.message.event.http.rss import NewFeedEvent
diff --git a/platypush/backend/mail/__init__.py b/platypush/backend/mail/__init__.py
index 84f1f8ab6b..5dd4474fec 100644
--- a/platypush/backend/mail/__init__.py
+++ b/platypush/backend/mail/__init__.py
@@ -9,9 +9,10 @@ from threading import Thread, RLock
 from typing import List, Dict, Any, Optional, Tuple
 
 from sqlalchemy import engine, create_engine, Column, Integer, String, DateTime
-from sqlalchemy.orm import sessionmaker, scoped_session, declarative_base
+from sqlalchemy.orm import sessionmaker, scoped_session
 
 from platypush.backend import Backend
+from platypush.common.db import declarative_base
 from platypush.config import Config
 from platypush.context import get_plugin
 from platypush.message.event.mail import (
@@ -40,6 +41,7 @@ class MailboxStatus(Base):
 
 # </editor-fold>
 
+
 # <editor-fold desc="Mailbox model">
 @dataclass
 class Mailbox:
diff --git a/platypush/common/db.py b/platypush/common/db.py
index 59be70308c..f1979e3b16 100644
--- a/platypush/common/db.py
+++ b/platypush/common/db.py
@@ -1,3 +1,10 @@
-from sqlalchemy.orm import declarative_base
+from sqlalchemy import __version__
+
+sa_version = tuple(map(int, __version__.split('.')))
+
+if sa_version >= (1, 4, 0):
+    from sqlalchemy.orm import declarative_base
+else:
+    from sqlalchemy.ext.declarative import declarative_base
 
 Base = declarative_base()
diff --git a/platypush/plugins/media/search/local.py b/platypush/plugins/media/search/local.py
index 76868aa4f3..21603dc1bc 100644
--- a/platypush/plugins/media/search/local.py
+++ b/platypush/plugins/media/search/local.py
@@ -12,9 +12,10 @@ from sqlalchemy import (
     PrimaryKeyConstraint,
     ForeignKey,
 )
-from sqlalchemy.orm import sessionmaker, scoped_session, declarative_base
+from sqlalchemy.orm import sessionmaker, scoped_session
 from sqlalchemy.sql.expression import func
 
+from platypush.common.db import declarative_base
 from platypush.config import Config
 from platypush.plugins.media import MediaPlugin
 from platypush.plugins.media.search import MediaSearcher

From 6fa179e769d50e6f219de05cc67d10bafef8b14e Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Mon, 24 Apr 2023 23:47:16 +0200
Subject: [PATCH 3/8] LINT fixes

---
 platypush/plugins/db/__init__.py | 35 ++++++++++++++++----------------
 1 file changed, 18 insertions(+), 17 deletions(-)

diff --git a/platypush/plugins/db/__init__.py b/platypush/plugins/db/__init__.py
index e92c2fe7e6..74b76eed66 100644
--- a/platypush/plugins/db/__init__.py
+++ b/platypush/plugins/db/__init__.py
@@ -45,7 +45,7 @@ class DbPlugin(Plugin):
     _db_error_wait_interval = 5.0
     _db_error_retries = 3
 
-    def __init__(self, engine=None, *args, **kwargs):
+    def __init__(self, *args, engine=None, **kwargs):
         """
         :param engine: Default SQLAlchemy connection engine string (e.g.
             ``sqlite:///:memory:`` or ``mysql://user:pass@localhost/test``)
@@ -86,16 +86,14 @@ class DbPlugin(Plugin):
         return self.engine
 
     @staticmethod
-    def _build_condition(table, column, value):  # type: ignore
-        if isinstance(value, str):
-            value = "'{}'".format(value)
-        elif not isinstance(value, int) and not isinstance(value, float):
-            value = "'{}'".format(str(value))
+    def _build_condition(table, column, value):  # pylint: disable=unused-argument
+        if isinstance(value, str) or not isinstance(value, (int, float)):
+            value = f"'{value}'"
 
-        return eval('table.c.{}=={}'.format(column, value))
+        return eval(f'table.c.{column}=={value}')  # pylint: disable=eval-used
 
     @action
-    def execute(self, statement, engine=None, *args, **kwargs):
+    def execute(self, statement, *args, engine=None, **kwargs):
         """
         Executes a raw SQL statement.
 
@@ -123,32 +121,34 @@ class DbPlugin(Plugin):
         with engine.connect() as connection:
             connection.execute(text(statement))
 
-    def _get_table(self, table, engine=None, *args, **kwargs):
+    def _get_table(self, table: str, engine=None, *args, **kwargs):
         if not engine:
             engine = self.get_engine(engine, *args, **kwargs)
 
         db_ok = False
         n_tries = 0
         last_error = None
+        table_ = None
 
         while not db_ok and n_tries < self._db_error_retries:
             try:
                 n_tries += 1
                 metadata = MetaData()
-                table = Table(table, metadata, autoload_with=engine)
+                table_ = Table(table, metadata, autoload_with=engine)
                 db_ok = True
             except Exception as e:
                 last_error = e
                 wait_time = self._db_error_wait_interval * n_tries
                 self.logger.exception(e)
-                self.logger.info('Waiting {} seconds before retrying'.format(wait_time))
+                self.logger.info('Waiting %s seconds before retrying', wait_time)
                 time.sleep(wait_time)
                 engine = self.get_engine(engine, *args, **kwargs)
 
         if not db_ok and last_error:
             raise last_error
 
-        return table, engine
+        assert table_, f'No such table: {table}'
+        return table_, engine
 
     @action
     def select(
@@ -385,8 +385,9 @@ class DbPlugin(Plugin):
 
         query = table.select().where(
             or_(
-                and_(
-                    self._build_condition(table, k, record.get(k)) for k in key_columns
+                and_(  # type: ignore
+                    self._build_condition(table, k, record.get(k))
+                    for k in key_columns  # type: ignore
                 )
                 for record in records
             )
@@ -520,11 +521,11 @@ class DbPlugin(Plugin):
 
         with engine.connect() as connection, conn_begin(connection):
             for record in records:
-                table, engine = self._get_table(table, engine=engine, *args, **kwargs)
-                delete = table.delete()
+                table_, engine = self._get_table(table, engine=engine, *args, **kwargs)
+                delete = table_.delete()
 
                 for k, v in record.items():
-                    delete = delete.where(self._build_condition(table, k, v))
+                    delete = delete.where(self._build_condition(table_, k, v))
 
                 connection.execute(delete)
 

From 37722d12cd76c93b3c8179e9dae51ef95cf665f2 Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Mon, 24 Apr 2023 23:55:50 +0200
Subject: [PATCH 4/8] No need for `session.begin` in `db.create_all`.

---
 platypush/plugins/db/__init__.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/platypush/plugins/db/__init__.py b/platypush/plugins/db/__init__.py
index 74b76eed66..b5a83df314 100644
--- a/platypush/plugins/db/__init__.py
+++ b/platypush/plugins/db/__init__.py
@@ -530,7 +530,7 @@ class DbPlugin(Plugin):
                 connection.execute(delete)
 
     def create_all(self, engine, base):
-        with self.get_session(engine, locked=True) as session, session.begin():
+        with self.get_session(engine, locked=True) as session:
             base.metadata.create_all(session.connection())
 
     @contextmanager

From e1cd22121a7198778536b7b30dd0d43f71a8790c Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Tue, 25 Apr 2023 10:31:49 +0200
Subject: [PATCH 5/8] Removed `connection.begin()` pattern from the `db`
 plugin.

SQLAlchemy should automatically begin a transaction on
connection/session creation. Plus, `.begin()` messes up things with
SQLAlchemy 2, which has `autobegin` enabled with no easy way of
disabling it.
---
 platypush/plugins/db/__init__.py | 67 ++++++++++++--------------------
 1 file changed, 24 insertions(+), 43 deletions(-)

diff --git a/platypush/plugins/db/__init__.py b/platypush/plugins/db/__init__.py
index b5a83df314..70708bd8f6 100644
--- a/platypush/plugins/db/__init__.py
+++ b/platypush/plugins/db/__init__.py
@@ -3,12 +3,7 @@ from contextlib import contextmanager
 from multiprocessing import RLock
 from typing import Optional, Generator, Union
 
-from sqlalchemy import (
-    create_engine,
-    Table,
-    MetaData,
-    __version__ as sa_version,
-)
+from sqlalchemy import create_engine, Table, MetaData
 from sqlalchemy.engine import Engine
 from sqlalchemy.exc import CompileError
 from sqlalchemy.orm import Session, sessionmaker, scoped_session
@@ -19,21 +14,6 @@ from platypush.plugins import Plugin, action
 session_locks = {}
 
 
-@contextmanager
-def conn_begin(conn):
-    """
-    Utility method to deal with `autobegin` being enabled on SQLAlchemy 2.0 but
-    not on earlier version.
-    """
-    sa_maj_ver = int(sa_version.split('.')[0])
-    if sa_maj_ver < 2:
-        yield conn.begin()
-    else:
-        yield conn._transaction
-
-    conn.commit()
-
-
 class DbPlugin(Plugin):
     """
     Database plugin. It allows you to programmatically select, insert, update
@@ -45,7 +25,7 @@ class DbPlugin(Plugin):
     _db_error_wait_interval = 5.0
     _db_error_retries = 3
 
-    def __init__(self, *args, engine=None, **kwargs):
+    def __init__(self, engine=None, **kwargs):
         """
         :param engine: Default SQLAlchemy connection engine string (e.g.
             ``sqlite:///:memory:`` or ``mysql://user:pass@localhost/test``)
@@ -62,7 +42,7 @@ class DbPlugin(Plugin):
 
         super().__init__()
         self.engine_url = engine
-        self.engine = self.get_engine(engine, *args, **kwargs)
+        self.engine = self.get_engine(engine, **kwargs)
 
     def get_engine(
         self, engine: Optional[Union[str, Engine]] = None, *args, **kwargs
@@ -116,12 +96,10 @@ class DbPlugin(Plugin):
             (see https:///docs.sqlalchemy.org/en/latest/core/engines.html)
         """
 
-        engine = self.get_engine(engine, *args, **kwargs)
-
-        with engine.connect() as connection:
+        with self.get_engine(engine, *args, **kwargs).connect() as connection:
             connection.execute(text(statement))
 
-    def _get_table(self, table: str, engine=None, *args, **kwargs):
+    def _get_table(self, table: str, *args, engine=None, **kwargs):
         if not engine:
             engine = self.get_engine(engine, *args, **kwargs)
 
@@ -344,17 +322,16 @@ class DbPlugin(Plugin):
                     connection, table, records, key_columns
                 )
 
-            with conn_begin(connection):
-                if insert_records:
-                    insert = table.insert().values(insert_records)
-                    ret = self._execute_try_returning(connection, insert)
-                    if ret:
-                        returned_records += ret
+            if insert_records:
+                insert = table.insert().values(insert_records)
+                ret = self._execute_try_returning(connection, insert)
+                if ret:
+                    returned_records += ret
 
-                if update_records and on_duplicate_update:
-                    ret = self._update(connection, table, update_records, key_columns)
-                    if ret:
-                        returned_records = ret + returned_records
+            if update_records and on_duplicate_update:
+                ret = self._update(connection, table, update_records, key_columns)
+                if ret:
+                    returned_records = ret + returned_records
 
         if returned_records:
             return returned_records
@@ -519,7 +496,7 @@ class DbPlugin(Plugin):
 
         engine = self.get_engine(engine, *args, **kwargs)
 
-        with engine.connect() as connection, conn_begin(connection):
+        with engine.connect() as connection:
             for record in records:
                 table_, engine = self._get_table(table, engine=engine, *args, **kwargs)
                 delete = table_.delete()
@@ -535,7 +512,7 @@ class DbPlugin(Plugin):
 
     @contextmanager
     def get_session(
-        self, engine=None, locked=False, autoflush=True, *args, **kwargs
+        self, *args, engine=None, locked=False, autoflush=True, **kwargs
     ) -> Generator[Session, None, None]:
         engine = self.get_engine(engine, *args, **kwargs)
         if locked:
@@ -544,16 +521,20 @@ class DbPlugin(Plugin):
             # Mock lock
             lock = RLock()
 
-        with lock, engine.connect() as conn, conn_begin(conn):
-            session = scoped_session(
+        with lock, engine.connect() as conn:
+            session_maker = scoped_session(
                 sessionmaker(
                     expire_on_commit=False,
                     autoflush=autoflush,
                 )
             )
 
-            session.configure(bind=conn)
-            yield session()
+            session_maker.configure(bind=conn)
+            session = session_maker()
+            yield session
+
+            session.flush()
+            session.commit()
 
 
 # vim:sw=4:ts=4:et:

From 4cc88fcf5fade49cc73b0b213f6ea9ea1dcfa93b Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Tue, 25 Apr 2023 10:35:12 +0200
Subject: [PATCH 6/8] Rewritten the `variable` plugin to use SQLAlchemy's ORM.

This removes the need for raw SQL statements and CREATE TABLE statements
that may be engine-specific.
---
 platypush/plugins/variable/__init__.py | 86 ++++++++++++++------------
 1 file changed, 46 insertions(+), 40 deletions(-)

diff --git a/platypush/plugins/variable/__init__.py b/platypush/plugins/variable/__init__.py
index 0346ba888a..9556b07f9d 100644
--- a/platypush/plugins/variable/__init__.py
+++ b/platypush/plugins/variable/__init__.py
@@ -1,6 +1,21 @@
-from platypush.config import Config
+from sqlalchemy import Column, String
+
+from platypush.common.db import declarative_base
 from platypush.context import get_plugin
 from platypush.plugins import Plugin, action
+from platypush.plugins.db import DbPlugin
+
+Base = declarative_base()
+
+
+# pylint: disable=too-few-public-methods
+class Variable(Base):
+    """Models the variable table"""
+
+    __tablename__ = 'variable'
+
+    name = Column(String, primary_key=True, nullable=False)
+    value = Column(String)
 
 
 class VariablePlugin(Plugin):
@@ -11,8 +26,6 @@ class VariablePlugin(Plugin):
     will be stored either persisted on a local database or on the local Redis instance.
     """
 
-    _variable_table_name = 'variable'
-
     def __init__(self, **kwargs):
         """
         The plugin will create a table named ``variable`` on the database
@@ -21,24 +34,14 @@ class VariablePlugin(Plugin):
         """
 
         super().__init__(**kwargs)
-        self.db_plugin = get_plugin('db')
-        self.redis_plugin = get_plugin('redis')
+        db_plugin = get_plugin('db')
+        redis_plugin = get_plugin('redis')
+        assert db_plugin, 'Database plugin not configured'
+        assert redis_plugin, 'Redis plugin not configured'
 
-        db = Config.get('db')
-        self.db_config = {
-            'engine': db.get('engine'),
-            'args': db.get('args', []),
-            'kwargs': db.get('kwargs', {})
-        }
-
-        self._create_tables()
-        # self._variables = {}
-
-    def _create_tables(self):
-        self.db_plugin.execute("""CREATE TABLE IF NOT EXISTS {}(
-            name varchar(255) not null primary key,
-            value text
-        )""".format(self._variable_table_name))
+        self.redis_plugin = redis_plugin
+        self.db_plugin: DbPlugin = db_plugin
+        self.db_plugin.create_all(self.db_plugin.get_engine(), Base)
 
     @action
     def get(self, name, default_value=None):
@@ -53,13 +56,10 @@ class VariablePlugin(Plugin):
         :returns: A map in the format ``{"<name>":"<value>"}``
         """
 
-        rows = self.db_plugin.select(table=self._variable_table_name,
-                                     filter={'name': name},
-                                     engine=self.db_config['engine'],
-                                     *self.db_config['args'],
-                                     **self.db_config['kwargs']).output
+        with self.db_plugin.get_session() as session:
+            var = session.query(Variable).filter_by(name=name).first()
 
-        return {name: rows[0]['value'] if rows else default_value}
+        return {name: (var.value if var is not None else default_value)}
 
     @action
     def set(self, **kwargs):
@@ -69,15 +69,24 @@ class VariablePlugin(Plugin):
         :param kwargs: Key-value list of variables to set (e.g. ``foo='bar', answer=42``)
         """
 
-        records = [{'name': k, 'value': v}
-                   for (k, v) in kwargs.items()]
+        with self.db_plugin.get_session() as session:
+            existing_vars = {
+                var.name: var
+                for var in session.query(Variable)
+                .filter(Variable.name.in_(kwargs.keys()))
+                .all()
+            }
 
-        self.db_plugin.insert(table=self._variable_table_name,
-                              records=records, key_columns=['name'],
-                              engine=self.db_config['engine'],
-                              on_duplicate_update=True,
-                              *self.db_config['args'],
-                              **self.db_config['kwargs'])
+            new_vars = {
+                name: Variable(name=name, value=value)
+                for name, value in kwargs.items()
+                if name not in existing_vars
+            }
+
+            for name, var in existing_vars.items():
+                var.value = kwargs[name]  # type: ignore
+
+            session.add_all([*existing_vars.values(), *new_vars.values()])
 
         return kwargs
 
@@ -90,12 +99,8 @@ class VariablePlugin(Plugin):
         :type name: str
         """
 
-        records = [{'name': name}]
-
-        self.db_plugin.delete(table=self._variable_table_name,
-                              records=records, engine=self.db_config['engine'],
-                              *self.db_config['args'],
-                              **self.db_config['kwargs'])
+        with self.db_plugin.get_session() as session:
+            session.query(Variable).filter_by(name=name).delete()
 
         return True
 
@@ -150,4 +155,5 @@ class VariablePlugin(Plugin):
 
         return self.redis_plugin.expire(name, expire)
 
+
 # vim:sw=4:ts=4:et:

From 440d70d9cfbee9e1900de394e81044c8b78a67e8 Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Tue, 25 Apr 2023 10:36:27 +0200
Subject: [PATCH 7/8] LINT/format fixes.

---
 platypush/user/__init__.py | 20 ++++++++------------
 1 file changed, 8 insertions(+), 12 deletions(-)

diff --git a/platypush/user/__init__.py b/platypush/user/__init__.py
index 7d268b592d..432774a42d 100644
--- a/platypush/user/__init__.py
+++ b/platypush/user/__init__.py
@@ -3,11 +3,11 @@ import datetime
 import hashlib
 import json
 import random
-import rsa
 import time
 from typing import Optional, Dict
 
 import bcrypt
+import rsa
 
 from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
 from sqlalchemy.orm import make_transient
@@ -73,7 +73,7 @@ class UserManager:
                 username=username,
                 password=self._encrypt_password(password),
                 created_at=datetime.datetime.utcnow(),
-                **kwargs
+                **kwargs,
             )
 
             session.add(record)
@@ -238,9 +238,7 @@ class UserManager:
             indent=None,
         )
 
-        return base64.b64encode(
-            rsa.encrypt(payload.encode('ascii'), pub_key)
-        ).decode()
+        return base64.b64encode(rsa.encrypt(payload.encode('ascii'), pub_key)).decode()
 
     def validate_jwt_token(self, token: str) -> Dict[str, str]:
         """
@@ -263,21 +261,19 @@ class UserManager:
 
         try:
             payload = json.loads(
-                rsa.decrypt(
-                    base64.b64decode(token.encode('ascii')),
-                    priv_key
-                ).decode('ascii')
+                rsa.decrypt(base64.b64decode(token.encode('ascii')), priv_key).decode(
+                    'ascii'
+                )
             )
         except (TypeError, ValueError) as e:
-            raise InvalidJWTTokenException(f'Could not decode JWT token: {e}')
+            raise InvalidJWTTokenException(f'Could not decode JWT token: {e}') from e
 
         expires_at = payload.get('expires_at')
         if expires_at and time.time() > expires_at:
             raise InvalidJWTTokenException('Expired JWT token')
 
         user = self.authenticate_user(
-            payload.get('username', ''),
-            payload.get('password', '')
+            payload.get('username', ''), payload.get('password', '')
         )
 
         if not user:

From dd60b8924d03e422020506864b084248f3a398ce Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Tue, 25 Apr 2023 10:41:37 +0200
Subject: [PATCH 8/8] Wrap the `PRAGMA` statement in `sqlalchemy.text`.

SQLAlchemy 2 no longer supports raw strings passed to `.execute()`
methods.
---
 platypush/plugins/entities/__init__.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/platypush/plugins/entities/__init__.py b/platypush/plugins/entities/__init__.py
index 443cf51ab2..74caf3cd5a 100644
--- a/platypush/plugins/entities/__init__.py
+++ b/platypush/plugins/entities/__init__.py
@@ -3,7 +3,7 @@ from threading import Thread
 from time import time
 from typing import Optional, Any, Collection, Mapping
 
-from sqlalchemy import or_
+from sqlalchemy import or_, text
 from sqlalchemy.orm import make_transient, Session
 
 from platypush.config import Config
@@ -198,7 +198,7 @@ class EntitiesPlugin(Plugin):
             if str(session.connection().engine.url).startswith('sqlite://'):
                 # SQLite requires foreign_keys to be explicitly enabled
                 # in order to proper manage cascade deletions
-                session.execute('PRAGMA foreign_keys = ON')
+                session.execute(text('PRAGMA foreign_keys = ON'))
 
             entities: Collection[Entity] = (
                 session.query(Entity).filter(Entity.id.in_(entities)).all()