8.25. SQLAlchemy Schema Reflection

  • SQLAlchemy allows for database reflection

SQLAlchemy tool Automap does the reflection of database and figures out how to do the mapping dynamically.

There is third-party tool sqlacodegen that generates the files (Python code) for you based on introspected tables. This is more robust solution.

Reflection

Loading Table objects based on reading from an existing database.

8.25.1. Reflection

"Reflection" refers to loading Table objects based on reading from an existing database. In order to create a reflection, first create empty metadata object:

>>> metadata2 = MetaData()

And then use it to introspect the database table:

>>> with engine.connect() as db:
...     astronauts = Table('astronauts', metadata2, autoload_with=db)

The reflected object is filled in with all the columns and constraints and is ready to use.

>>> print(astronauts.c)
<sqlalchemy.sql.base.ImmutableColumnCollection object at 0x...>

>>> print(astronauts.primary_key)
PrimaryKeyConstraint(Column('id', INTEGER(), table=<astronaut>, primary_key=True, nullable=False))

>>> print(select(astronauts))
SELECT astronaut.id, astronaut.firstname, astronaut.lastname, astronaut.birthdate, astronaut.height, astronaut.weight, astronaut.agency
FROM astronaut

8.25.2. Inspection

Information about a database at a more specific level is available using the Inspector object. Inspector will work with an engine or a connection. [1]

First import the inspector:

>>> from sqlalchemy import inspect

Attach it to the engine:

>>> inspector = inspect(engine)

You can query the database to get all tables:

>>> inspector.get_table_names()
['astronaut']

Or get information about columns:

>>> inspector.get_columns('astronaut')
[{'name': 'id', 'type': INTEGER(), 'nullable': False, 'default': None, 'autoincrement': 'auto', 'primary_key': 1},
 {'name': 'firstname', 'type': VARCHAR(length=50), 'nullable': False, 'default': None, 'autoincrement': 'auto', 'primary_key': 0},
 {'name': 'lastname', 'type': VARCHAR(length=50), 'nullable': False, 'default': None, 'autoincrement': 'auto', 'primary_key': 0},
 {'name': 'birthdate', 'type': DATE(), 'nullable': True, 'default': None, 'autoincrement': 'auto', 'primary_key': 0},
 {'name': 'height', 'type': INTEGER(), 'nullable': True, 'default': None, 'autoincrement': 'auto', 'primary_key': 0},
 {'name': 'weight', 'type': NUMERIC(precision=3, scale=2), 'nullable': True, 'default': None, 'autoincrement': 'auto', 'primary_key': 0},
 {'name': 'agency', 'type': VARCHAR(length=9), 'nullable': True, 'default': None, 'autoincrement': 'auto', 'primary_key': 0}]

Or constraints:

>>> inspector.get_foreign_keys('astronaut')
[]

Currently supported constraints:

  • FOREIGNKEY

  • UNIQUE

  • CHECK

Currently not supported:

  • Functional Indexes (PostgreSQL)

  • EXCLUDE (PostgreSQL)

8.25.3. Reflecting an Entire Schema

The MetaData object also includes a feature that will reflect all the tables in particular schema at once. [1]

>>> metadata3 = MetaData()
>>>
>>> with engine.connect() as db:
...     metadata3.reflect(db)

Note, that this will produce a lot of database queries. The Tables objects are then in the metadata.tables collection:

>>> metadata3.tables
FacadeDict({
    'published': Table('published', MetaData(),
                    Column('pub_id', INTEGER(), table=<published>, primary_key=True, nullable=False),
                    Column('pub_timestamp', Date(), table=<published>),
                    Column('story_id', INTEGER(), ForeignKey('story.story_id'), table=<published>),
                    Column('version_id', INTEGER(), ForeignKey('story.version_id'), table=<published>), schema=None),
    'story': Table('story', MetaData(),
                    Column('story_id', INTEGER(), table=<story>, primary_key=True, nullable=False),
                    Column('version_id', INTEGER(), table=<story>, primary_key=True, nullable=False),
                    Column('headline', VARCHAR(length=100), table=<story>, nullable=False),
                    Column('body', TEXT(), table=<story>), schema=None),
    'users': Table('users', MetaData(),
                    Column('uid', INTEGER(), table=<users>, primary_key=True, nullable=False),
                    Column('firstname', VARCHAR(), table=<users>, nullable=False),
                    Column('lastname', VARCHAR(), table=<users>, nullable=False), schema=None)})

>>> story = metadata3.tables['story']
>>> published = metadata3.tables['published']

>>> story
Table('story', MetaData(),
      Column('story_id', INTEGER(), table=<story>, primary_key=True, nullable=False),
      Column('version_id', INTEGER(), table=<story>, primary_key=True, nullable=False),
      Column('headline', VARCHAR(length=100), table=<story>, nullable=False),
      Column('body', TEXT(), table=<story>), schema=None)

>>> published
Table('published', MetaData(),
      Column('pub_id', INTEGER(), table=<published>, primary_key=True, nullable=False),
      Column('pub_timestamp', DATE(), table=<published>),
      Column('story_id', INTEGER(), ForeignKey('story.story_id'), table=<published>),
      Column('version_id', INTEGER(), ForeignKey('story.version_id'), table=<published>), schema=None)

This is useful if you have an existing database and you want to write queries against it.

>>> query = select(story).join(published)
>>> print(query)
SELECT story.story_id, story.version_id, story.headline, story.body
FROM story JOIN published ON story.story_id = published.story_id AND story.version_id = published.version_id

8.25.4. References