How to use SQLAlchemy models to generate raw SQL which is able to create all tables?

Question

It is possible to generate raw SQL statement to create any SQLAlchemy table as shown here.

However, I would like to generate raw SQL which creates all tables, not just one.

Let's assume I have such tables (this table design is not real-world, but it shows the issue well):

import sqlalchemy
from sqlalchemy import Column, Integer, Text, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import CreateTable
from sqlalchemy.dialects import postgresql

engine = sqlalchemy.create_engine('postgresql://postgres:postgres@127.0.0.1:5432/postgres', echo=True)
Base = declarative_base()


class Product(Base):
    __tablename__ = "product"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(Text, nullable=True)
    user_id = Column(Integer, ForeignKey('user.id'), nullable=True)

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(Text, nullable=True)
    product_id = Column(Integer, ForeignKey('product.id'), nullable=True)


open("create_tables.sql", 'w').close() # to clear the file

for table in Base.metadata.tables.values():
    table_sql = str(CreateTable(table).compile(dialect=postgresql.dialect()))
    with open("create_tables.sql", "a") as myfile:
        myfile.write(table_sql)


Base.metadata.create_all(engine)

(this is complete program, you can run it as-is)

In code above you can see that I use for loop to generate raw SQL:

for table in Base.metadata.tables.values():
    table_sql = str(CreateTable(table).compile(dialect=postgresql.dialect()))
    with open("create_tables.sql", "a") as myfile:
        myfile.write(table_sql)

Resulting file contains:

CREATE TABLE product (
    id SERIAL NOT NULL, 
    name TEXT, 
    user_id INTEGER, 
    PRIMARY KEY (id), 
    FOREIGN KEY(user_id) REFERENCES "user" (id)
)

CREATE TABLE "user" (
    id SERIAL NOT NULL, 
    name TEXT, 
    product_id INTEGER, 
    PRIMARY KEY (id), 
    FOREIGN KEY(product_id) REFERENCES product (id)
)

But I can't execute this SQL code, it will give me error due to circular reference. (and I need to add at least some ;)

---

SQLAlchemy can however create those tables without problem when I use:

Base.metadata.create_all(engine)

With echo=True the generated SQL is visible in log:

2021-10-18 05:37:24,442 INFO sqlalchemy.engine.Engine select version()
2021-10-18 05:37:24,443 INFO sqlalchemy.engine.Engine [raw sql] {}
2021-10-18 05:37:24,447 INFO sqlalchemy.engine.Engine select current_schema()
2021-10-18 05:37:24,447 INFO sqlalchemy.engine.Engine [raw sql] {}
2021-10-18 05:37:24,449 INFO sqlalchemy.engine.Engine show standard_conforming_strings
2021-10-18 05:37:24,450 INFO sqlalchemy.engine.Engine [raw sql] {}
2021-10-18 05:37:24,454 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2021-10-18 05:37:24,455 INFO sqlalchemy.engine.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s     
2021-10-18 05:37:24,456 INFO sqlalchemy.engine.Engine [generated in 0.00073s] {'name': 'product'}  
2021-10-18 05:37:24,459 INFO sqlalchemy.engine.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s     
2021-10-18 05:37:24,460 INFO sqlalchemy.engine.Engine [cached since 0.004542s ago] {'name': 'user'}2021-10-18 05:37:24,463 INFO sqlalchemy.engine.Engine
CREATE TABLE product (
        id SERIAL NOT NULL,
        name TEXT,
        user_id INTEGER,
        created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
        PRIMARY KEY (id)
)


2021-10-18 05:37:24,464 INFO sqlalchemy.engine.Engine [no key 0.00116s] {}
2021-10-18 05:37:24,480 INFO sqlalchemy.engine.Engine
CREATE TABLE "user" (
        id SERIAL NOT NULL,
        name TEXT,
        product_id INTEGER,
        created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
        PRIMARY KEY (id)
)


2021-10-18 05:37:24,481 INFO sqlalchemy.engine.Engine [no key 0.00128s] {}
2021-10-18 05:37:24,494 INFO sqlalchemy.engine.Engine ALTER TABLE product ADD FOREIGN KEY(user_id) 
REFERENCES "user" (id)
2021-10-18 05:37:24,495 INFO sqlalchemy.engine.Engine [no key 0.00053s] {}
2021-10-18 05:37:24,497 INFO sqlalchemy.engine.Engine ALTER TABLE "user" ADD FOREIGN KEY(product_id) REFERENCES product (id)
2021-10-18 05:37:24,497 INFO sqlalchemy.engine.Engine [no key 0.00047s] {}
2021-10-18 05:37:24,500 INFO sqlalchemy.engine.Engine COMMIT

My question is how to get this SQL, which is visible in the log above and is able to create all tables in one transaction, as Python string (or .sql file)?

I want to be able to check it / review it and then apply it (or part of it) manually to production server.

Answer 1

The create_mock_engine function, introduced in SQLAlchemy v1.4^* will generate the required CREATE TABLE statements, and moreover will generate the foreign key constraints in separate ALTER TABLE statements, thus avoiding problems with circular references.

create_mock_engine takes two arguments, a minimal database URL and a function which receives a compilable object. The function is called for each statement, so we can append statement delimiters before emitting output.

This is the complete fixed code for the question:

import pathlib

import sqlalchemy as sa 
from sqlalchemy import orm
from sqlalchemy import Column, Integer, Text, ForeignKey


class Base(orm.DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "product"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(Text, nullable=True)
    user_id = Column(Integer, ForeignKey('user.id'), nullable=True)

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(Text, nullable=True)
    product_id = Column(Integer, ForeignKey('product.id'), nullable=True)


path = pathlib.Path('so69610811.sql')
path.unlink()

with path.open(mode='a') as f:

    def dump(sql, *multiparams, **params):
        stmt = sql.compile(dialect=engine.dialect)
        # Append the statement delimiter, and a newline for legibility.
        f.write(f'{stmt};\n')

    engine = sa.create_mock_engine('postgresql+psycopg2://', dump)
    Base.metadata.create_all(engine, checkfirst=False)

and this is the generated SQL:

CREATE TABLE product (
    id SERIAL NOT NULL, 
    name TEXT, 
    user_id INTEGER, 
    PRIMARY KEY (id)
)

;

CREATE TABLE "user" (
    id SERIAL NOT NULL, 
    name TEXT, 
    product_id INTEGER, 
    PRIMARY KEY (id)
)

;
ALTER TABLE "user" ADD FOREIGN KEY(product_id) REFERENCES product (id);
ALTER TABLE product ADD FOREIGN KEY(user_id) REFERENCES "user" (id);

---

^* for earlier versions use engine = create_engine(url, strategy='mock', executor=dump)

Answer 2

This seems to be one way to get DDL for all tables:

from sqlalchemy.dialects import postgresql
from sqlalchemy.schema import CreateTable


# Base is the declarative base
for table in Base.metadata.sorted_tables:
    print(CreateTable(table).compile(dialect=postgresql.dialect()))

Answer 3

You may do this:

1. Generate DDL of the first type into a file;
2. Cut/paste statements like FOREIGN KEY(user_id) REFERENCES "user" (id) to the end of the file;
3. Edit them to become ALTER TABLE product ADD FOREIGN KEY(user_id) REFERENCES "user" (id) (like the second script);

So you first create the tables and then add the foreign key constraints. These are safe & trivial changes that keep the logic intact. Your example becomes:

CREATE TABLE product (
    id SERIAL NOT NULL, 
    name TEXT, 
    user_id INTEGER, 
    PRIMARY KEY (id) 
    -- cut FOREIGN KEY(user_id) REFERENCES "user" (id) and remove a comma
);

CREATE TABLE "user" (
    id SERIAL NOT NULL, 
    name TEXT, 
    product_id INTEGER, 
    PRIMARY KEY (id)
    -- Cut FOREIGN KEY(product_id) REFERENCES product (id) and remove a comma
);

-- Paste and edit here. Prepend "alter table <tablename> add "
alter table product add FOREIGN KEY(user_id) REFERENCES "user" (id);
alter table "user" add FOREIGN KEY(product_id) REFERENCES product (id);

Please note added semicolons after statements. They are missing in the generated DDL.