Don't pass kwargs to connection.execute()

Resolve the following RemovedIn20Warning warning:

  The connection.execute() method in SQLAlchemy 2.0 will accept
  parameters as a single dictionary or a single sequence of dictionaries
  only. Parameters passed as keyword arguments, tuples or positionally
  oriented dictionaries and/or tuples will no longer be accepted.

Change-Id: I44675fce86337696b6494abc03e8058af32686c6
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2021-07-19 11:36:01 +01:00
parent 62d77fc6df
commit e1039e0849
4 changed files with 9 additions and 11 deletions

View File

@ -605,9 +605,11 @@ class PostgresqlBackendImpl(BackendImpl):
return bool(
engine.scalar(
sqlalchemy.text(
"SELECT datname FROM pg_database "
"WHERE datname=:name"), name=ident)
"SELECT datname FROM pg_database WHERE datname=:name"
),
{'name': ident},
)
)
def _close_out_database_users(self, conn, ident):
"""Attempt to guarantee a database can be dropped.
@ -631,7 +633,9 @@ class PostgresqlBackendImpl(BackendImpl):
"WHERE usename=current_user AND "
"pid != pg_backend_pid() AND "
"datname=:dname"
), dname=ident)
),
{'dname': ident},
)
def _random_ident():

View File

@ -1118,7 +1118,7 @@ def get_non_innodb_tables(connectable, skip_tables=('migrate_version',
params['database'] = connectable.engine.url.database
query = text(query_str)
noninnodb = connectable.execute(query, **params)
noninnodb = connectable.execute(query, params)
return [i[0] for i in noninnodb]

View File

@ -52,11 +52,6 @@ class WarningsFixture(fixtures.Fixture):
message=r'The Engine.execute\(\) method is considered legacy .*',
category=sqla_exc.SADeprecationWarning)
warnings.filterwarnings(
'once',
message=r'The connection.execute\(\) method in SQLAlchemy 2.0 .*',
category=sqla_exc.SADeprecationWarning)
warnings.filterwarnings(
'once',
message=r'Calling the mapper\(\) function directly outside .*',

View File

@ -324,8 +324,7 @@ class MySQLModeTestCase(db_test_base._MySQLOpportunisticTestCase):
def _test_string_too_long(self, value):
with self.connection.begin():
self.connection.execute(self.test_table.insert(),
bar=value)
self.connection.execute(self.test_table.insert(), {'bar': value})
result = self.connection.execute(self.test_table.select())
return result.fetchone().bar