Is there a way to bind a session object to MetaData.create_all method in SQLAlchemy?
<p>I have an application that creates tables on demand using <strong>SQLAlchemy</strong>. More precisely <strong>Flask-SQLAlchemy</strong> and <strong>PostgreSQL</strong> as database. </p>

<p>To do that, <strong>(1)</strong> I create a PostgreSQL schema to hold the new tables:</p>

<pre><code># extra checks on the schema name before I execute it.

statement = 'CREATE SCHEMA IF NOT EXISTS {}'.format(schema_name)
database.session.execute(statement)
database.session.commit()
</code></pre>

<p>After that, <strong>(2)</strong> I change PostgreSQL <code>search_path</code> value to the schema that I have created.</p>

<p><strong>(3)</strong> and then I make a list of tables that I want to create in the database and pass it to <code>create_all</code> MetaData's method:</p>

<pre><code>metadata.create_all(database.engine, tables=list_of_tables)
</code></pre>

<p><strong>SQLAlchemy</strong> make queries to check if those tables already exist (on the new schema) and then send <code>CREATE TABLE</code> statements to the database.</p>

<p>The tables are created correctly, in the desired schema and everything works Ok.</p>

<p>My problem starts when I wrap all those tasks in a <strong>nested transaction</strong> (using PostgreSQL SAVEPOINTs) for <strong>testing purpose</strong>, in order to rollback everything in the current session in the end of the tests. I am using the example in <a href="http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites" rel="nofollow">Supporting Tests with Rollbacks</a>, from the SQLAlchemy documentation.</p>

<p>The schema creation happen inside the nested transaction, but <code>MetaData.crate_all</code> make the job in another transaction, and is unable to find the new schema in the database, because the new schema is just alive in the wrapped session and not physically created in the database.</p>

<p>This make the tests fail with <code>(psycopg2.ProgrammingError) no schema has been selected to create in ...</code></p>

<p>The solution that I was thinking is to create the tables one by one using the wrapped session, or figuring out how to bind create_all to the wrapped session.</p>

<h2>Update</h2>

<p>To clarify more my question, as I say in the top, the application must create tables in the database on <strong>demand</strong> inside a new schema on the go. This means that I can't set my declarative table with a fixed schema. Because we don't know what will be the schema name, and in consequence, to what schema the table will belong.</p>

<p><a href="http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites" rel="nofollow">"Joining a Session into an External Transaction (such as for test suites)"</a> from the SQLAlchemy docs is a good starting point in this case. I would refactor your approach a bit though: create the schema in a transaction of a connection acquired from your <code>engine</code>. Then join the test session to said connection, perform your tests and rollback. Here's a quick example:</p>

<pre><code>In [2]: engine
Out[2]: Engine(postgresql://baz@localhost/sopython)

In [3]: conn = engine.connect()

In [4]: trans = conn.begin()

In [5]: class Foo(Base):
   ...:     __tablename__ = 'foo'
   ...:     __table_args__ = {'schema': 'bar'}
   ...:     id = Column(Integer, primary_key=True)

In [7]: from sqlalchemy.schema import CreateSchema

In [8]: conn.execute(CreateSchema('bar'))
Out[8]: &lt;sqlalchemy.engine.result.ResultProxy at 0x7f8fd4084d68&gt;

In [9]: Base.metadata.create_all(conn)  # Explicitly pass `conn` as bind!

In [10]: session = Session(bind=conn)  # This here joins the `session` to the
    ...:                               # external transaction.
    ...:

In [11]: session.query(Foo).all()
Out[11]: []

In [12]: trans.rollback()  # Undo everything.

In [13]: session.query(Foo).all()  # Table should not exist anymore.
---------------------------------------------------------------------------
ProgrammingError                          Traceback (most recent call last)
  ...
ProgrammingError: (psycopg2.ProgrammingError) relation "bar.foo" does not exist
LINE 2: FROM bar.foo
             ^
 [SQL: 'SELECT bar.foo.id AS bar_foo_id \nFROM bar.foo']

In [14]: Base.metadata.create_all()  # Uses metadata.engine implicitly,
    ...:                             # acquires a new connection etc, but
    ...:                             # the schema is now gone.
    ...:
---------------------------------------------------------------------------
ProgrammingError                          Traceback (most recent call last)
  ...
ProgrammingError: (psycopg2.ProgrammingError) schema "bar" does not exist
 [SQL: '\nCREATE TABLE bar.foo (\n\tid SERIAL NOT NULL, \n\tPRIMARY KEY (id)\n)\n\n']
</code></pre>

<p>The session is clever enough to use nested transactions (savepoints) automatically when it is passed an existing connection with an open transaction as bind. If your session has to rollback in a test, see the topic "Supporting Tests with Rollbacks" at the bottom of the linked documentation.</p>

<p>Since you're using <a href="/questions/tagged/flask" class="post-tag" title="show questions tagged &#39;flask&#39;" rel="tag">flask</a> and <a href="/questions/tagged/flask-sqlalchemy" class="post-tag" title="show questions tagged &#39;flask-sqlalchemy&#39;" rel="tag">flask-sqlalchemy</a>, you may have to adapt this a bit to fit your testing environment. <a href="http://alexmic.net/flask-sqlalchemy-pytest/" rel="nofollow">This post</a> by Alex Michael has an example for flask-sqlalchemy and <a href="http://doc.pytest.org/en/latest/" rel="nofollow">pytest</a>. The gist of it is to create a new joined session during setup:</p>

<pre><code>connection = db.engine.connect()
transaction = connection.begin()

options = dict(bind=connection, binds={})
session = db.create_scoped_session(options=options)

db.session = session
</code></pre>

<p>and to do the required rollbacks, closing etc. during teardown:</p>

<pre><code>def teardown():
    transaction.rollback()
    connection.close()
    session.remove()
</code></pre>

python postgresql sqlalchemy