openstack/taskflow
Code Issues Proposed changes

Replace deprecated Engine.execute

Browse Source 
Engine.execute was remvoed in SQLAlchemy 2.0.

Change-Id: I3a3e3263c54742ba793b293a2358f6e8202c21bb
This commit is contained in:
Takashi Kajinami 2024-04-15 14:17:04 +09:00
parent ff45bd3d2d
commit bfac5a7489
1 changed files with 25 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
tools/schema_generator.py
View File

@ -17,6 +17,7 @@
import contextlib import contextlib
import re import re
import sqlalchemy as sa
import tabulate import tabulate
from taskflow.persistence.backends import impl_sqlalchemy from taskflow.persistence.backends import impl_sqlalchemy
@ -53,29 +54,30 @@ def main():
with contextlib.closing(backend.get_connection()) as conn: with contextlib.closing(backend.get_connection()) as conn:
conn.upgrade() conn.upgrade()
# Now make a prettier version of that schema... # Now make a prettier version of that schema...
tables = backend.engine.execute(TABLE_QUERY) with backend.engine.connect() as conn, conn.begin():
table_names = [r[0] for r in tables] tables = conn.execute(sa.text(TABLE_QUERY))
for i, table_name in enumerate(table_names): table_names = [r[0] for r in tables]
pretty_name = NAME_MAPPING.get(table_name, table_name) for i, table_name in enumerate(table_names):
print("*" + pretty_name + "*") pretty_name = NAME_MAPPING.get(table_name, table_name)
# http://www.sqlite.org/faq.html#q24 print("*" + pretty_name + "*")
table_name = table_name.replace("\"", "\"\"") # http://www.sqlite.org/faq.html#q24
rows = [] table_name = table_name.replace("\"", "\"\"")
for r in backend.engine.execute(SCHEMA_QUERY % table_name): rows = []
# Cut out the numbers from things like VARCHAR(12) since for r in conn.execute(sa.text(SCHEMA_QUERY % table_name)):
# this is not very useful to show users who just want to # Cut out the numbers from things like VARCHAR(12) since
# see the basic schema... # this is not very useful to show users who just want to
row_type = re.sub(r"\(.*?\)", "", r['type']).strip() # see the basic schema...
if not row_type: row_type = re.sub(r"\(.*?\)", "", r['type']).strip()
raise ValueError("Row %s of table '%s' was empty after" if not row_type:
" cleaning" % (r['cid'], table_name)) raise ValueError("Row %s of table '%s' was empty after"
rows.append([r['name'], row_type, to_bool_string(r['pk'])]) " cleaning" % (r['cid'], table_name))
contents = tabulate.tabulate( rows.append([r['name'], row_type, to_bool_string(r['pk'])])
rows, headers=['Name', 'Type', 'Primary Key'], contents = tabulate.tabulate(
tablefmt="rst") rows, headers=['Name', 'Type', 'Primary Key'],
print("\n%s" % contents.strip()) tablefmt="rst")
if i + 1 != len(table_names): print("\n%s" % contents.strip())
print("") if i + 1 != len(table_names):
print("")
if __name__ == '__main__': if __name__ == '__main__':