Refreshing the inspector in SQLAlchemy

This code runs every time a user starts up their server. It's meant to smooth out the process.

# Checking if the database has tables
inspector = inspect(engine)
tables = inspector.get_table_names()
print(tables)  # Should be "[]"

if not (tables):
    # There are no tables in the database.
    db.create_all()
    db.session.commit()
    print("Created database tables.")

del inspector

Next, we want to check every table and see if it's empty. If it is, then we print the name of the table.

with engine.connect() as connection:
    # Iterating through all the tables in the database...
    inspector = inspect(engine)
    tables = inspector.get_table_names()
    print(tables) # Should return something like ['users', 'posts'] 

    for table_name in tables:
        # Does this table have any rows?
        result = connection.execute(
            "SELECT COUNT(*) FROM {}".format(table_name))
        for row in result:
            if row[0] == 0:
                # This table doesn't have any rows.
                # So we fill it...
    del inspector

This code executes chronologically as shown.

Is there a better way of executing this procedure that doesn't require me to delete the inspector object and recreate it? I have a feeling this is not best practice.

Since SQLAlchemy version 2.0 engine inspectors expose clear_cache method which, if called, will force the inspector to query the database to fetch information about the schema.

import atexit
import random

import sqlalchemy as sa

engine = sa.create_engine('sqlite://', echo=False)
atexit.register(engine.dispose)

metadata = sa.MetaData()

for n in range(1, 4):
    sa.Table(f't{n}', metadata, sa.Column('col', sa.Integer))

metadata.create_all(engine)

insp = sa.inspect(engine)

print(f'{insp.get_table_names() = }')
tbl = metadata.tables[random.choice(insp.get_table_names())]
tbl.drop(engine)
print(f'{insp.get_table_names() = }')
insp.clear_cache()
print(f'{insp.get_table_names() = }')

Output:

insp.get_table_names() = ['t1', 't2', 't3']
insp.get_table_names() = ['t1', 't2', 't3']
insp.get_table_names() = ['t2', 't3']

Some observations on the code in the question:

• It isn't necessary to condition calling db.create_all() on the content of get_table_names: create_all will automatically create any tables that don't exist already.

• If you have a lot of tables, or the tables contain a lot of records then looping over the tables and counting the records in each may be slow

  • To avoid looping, create a union query that combines all the queries in one
  • To avoid the overhead of counting all the records, consider that you only need to know whether a table contains any records. So assuming that each table has a primary key defined (or some other indexed, numeric, not nullable column), query for the column's minimum or maximum value, which can be read quickly from the index.
  • The resulting code might look like this:

  import functools
  ...
  tables = [metadata.tables[t] for t in insp.get_table_names()]
  stmt = functools.reduce(
      sa.union,
     [sa.select(sa.literal(tbl.name), sa.func.min(tbl.c.id)) for tbl in tables],
  )
  with engine.connect() as conn:
      rows = conn.execute(stmt)
      for table_name, result in rows:
          if result is None:
              # Populate the table