Getting SQLAlchemy to issue CREATE SCHEMA on create_all

I wrote a function that creates the declared schemas based on the accepted answer. It uses the schema value from the __table_args__ dict from each mapped class.

from sqlalchemy import event, DDL


# Import or write your mapped classes and configuration here

def init_db():
    for mapper in Base.registry.mappers:
        cls = mapper.class_
        if issubclass(cls, Base):
            table_args = getattr(cls, '__table_args__', None)
            if table_args:
                schema = table_args.get('schema')
                if schema:
                    stmt = f"CREATE SCHEMA IF NOT EXISTS {schema}"
                    event.listen(Base.metadata, 'before_create', DDL(stmt))
    Base.metadata.create_all(bind=engine)

I have done it manually on my db init script like so:

from sqlalchemy.schema import CreateSchema
engine.execute(CreateSchema('my_schema'))

But this seems less magical than I was expecting.

I ran into the same issue and believe the "cleanest" way of issuing the DDL is something like this:

from sqlalchemy import event
from sqlalchemy.schema import CreateSchema

event.listen(Base.metadata, 'before_create', CreateSchema('my_schema'))

This will ensure that before anything contained in the metadata of your base is created, you have the schema for it. This does, however, not check if the schema already exists.

You can do CreateSchema('my_schema').execute_if(callback_=check_schema) if you can be bothered to write the check_schema callback ("Controlling DDL Sequences" on should_create in docs). Or, as an easy way out, just use DDL("CREATE SCHEMA IF NOT EXISTS my_schema") instead (for Postgres):

from sqlalchemy import DDL

event.listen(Base.metadata, 'before_create', DDL("CREATE SCHEMA IF NOT EXISTS my_schema"))