sqlalchemy.py 文件源码

def test_connection(
    ctx: object, metadata: MetaData, engine: Engine,
    real_transaction: bool=False
) -> typing.Generator:
    """Joining a SQLAlchemy session into an external transaction for test suit.

    .. seealso::

       Documentation of the SQLAlchemy session used in test suites.
          <http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites>

    """  # noqa
    if real_transaction:
        metadata.create_all(engine)
        try:
            yield engine
        finally:
            metadata.drop_all(engine, checkfirst=True)
        return
    connection = engine.connect()
    try:
        metadata.drop_all(connection, checkfirst=True)
        transaction = connection.begin()
        try:
            metadata.create_all(bind=connection)
            ctx._test_fx_connection = connection
            try:
                yield connection
            finally:
                del ctx._test_fx_connection
        finally:
            transaction.rollback()
    finally:
        connection.close()
    engine.dispose()