conftest.py 文件源码

def fx_sess(fx_engine):
    metadata = Base.metadata
    foreign_key_turn_off = {
        'mysql': 'SET FOREIGN_KEY_CHECKS=0;',
        'postgresql': 'SET CONSTRAINTS ALL DEFERRED;',
        'sqlite': 'PRAGMA foreign_keys = OFF;',
    }
    foreign_key_turn_on = {
        'mysql': 'SET FOREIGN_KEY_CHECKS=1;',
        'postgresql': 'SET CONSTRAINTS ALL IMMEDIATE;',
        'sqlite': 'PRAGMA foreign_keys = ON;',
    }
    truncate_query = {
        'mysql': 'TRUNCATE TABLE {};',
        'postgresql': 'TRUNCATE TABLE {} RESTART IDENTITY CASCADE;',
        'sqlite': 'DELETE FROM {};',
    }
    error = False
    with fx_engine.begin() as conn:
        try:
            conn.execute(foreign_key_turn_off[fx_engine.name])
        except ProgrammingError:
            error = True

        for table in reversed(metadata.sorted_tables):
            try:
                conn.execute(truncate_query[fx_engine.name].format(table.name))
            except ProgrammingError:
                error = True

        try:
            conn.execute(foreign_key_turn_on[fx_engine.name])
        except ProgrammingError:
            error = True

    if error:
        metadata = Base.metadata
        metadata.drop_all(bind=fx_engine)
        metadata.create_all(bind=fx_engine)

    sess = Session(bind=fx_engine)
    yield sess
    sess.rollback()