revil-O/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

[db] Wrap SQL statements into connection.begin() blocks.

Browse source 
The latest release of SQLAlchemy 2.x has apparently removed the
`autocommit` implicit logic for good.

Mutations should be explicitly wrapped into `with ... begin()` blocks,
or they will be rolled back when the connection is closed.
This commit is contained in:
Fabio Manganiello 2024-11-03 10:38:22 +01:00
parent 1cb42f8923
commit dd02be12fc
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
1 changed files with 6 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
platypush/plugins/db/__init__.py
View file

@ -105,7 +105,9 @@ class DbPlugin(Plugin):
(see https:///docs.sqlalchemy.org/en/latest/core/engines.html) (see https:///docs.sqlalchemy.org/en/latest/core/engines.html)
""" """
with self.get_engine(engine, *args, **kwargs).connect() as connection: with self.get_engine(
engine, *args, **kwargs
).connect() as connection, connection.begin():
connection.execute(text(statement)) connection.execute(text(statement))
def _get_table(self, table: str, *args, engine=None, **kwargs): def _get_table(self, table: str, *args, engine=None, **kwargs):
@ -324,7 +326,7 @@ class DbPlugin(Plugin):
update_records = [] update_records = []
returned_records = [] returned_records = []
with engine.connect() as connection: with engine.connect() as connection, connection.begin():
# Upsert case # Upsert case
if key_columns: if key_columns:
insert_records, update_records = self._get_new_and_existing_records( insert_records, update_records = self._get_new_and_existing_records(
@ -462,7 +464,7 @@ class DbPlugin(Plugin):
} }
""" """
engine = self.get_engine(engine, *args, **kwargs) engine = self.get_engine(engine, *args, **kwargs)
with engine.connect() as connection: with engine.connect() as connection, connection.begin():
table, engine = self._get_table(table, *args, engine=engine, **kwargs) table, engine = self._get_table(table, *args, engine=engine, **kwargs)
return self._update(connection, table, records, key_columns) return self._update(connection, table, records, key_columns)
@ -505,7 +507,7 @@ class DbPlugin(Plugin):
engine = self.get_engine(engine, *args, **kwargs) engine = self.get_engine(engine, *args, **kwargs)
with engine.connect() as connection: with engine.connect() as connection, connection.begin():
for record in records: for record in records:
table_, engine = self._get_table(table, *args, engine=engine, **kwargs) table_, engine = self._get_table(table, *args, engine=engine, **kwargs)
delete = table_.delete() delete = table_.delete()