Skip to content

Instantly share code, notes, and snippets.

@absent1706
Last active April 5, 2023 03:50
Show Gist options
  • Save absent1706/3ccc1722ea3ca23a5cf54821dbc813fb to your computer and use it in GitHub Desktop.
Save absent1706/3ccc1722ea3ca23a5cf54821dbc813fb to your computer and use it in GitHub Desktop.
Sqlalchemy: Truncate all tables
def truncate_db(engine):
# delete all table data (but keep tables)
# we do cleanup before test 'cause if previous test errored,
# DB can contain dust
meta = MetaData(bind=engine, reflect=True)
con = engine.connect()
trans = con.begin()
con.execute('SET FOREIGN_KEY_CHECKS = 0;')
for table in meta.sorted_tables:
con.execute(table.delete())
con.execute('SET FOREIGN_KEY_CHECKS = 1;')
trans.commit()
@VVelda
Copy link

VVelda commented Jun 3, 2021

It may be enough to disable a foreign key checks just for the current session:
con.execute('SET SESSION FOREIGN_KEY_CHECKS = ON')

@photonq2
Copy link

photonq2 commented Feb 6, 2023

from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import config

from sqlalchemy import MetaData, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base


config = config.Config("alembic.ini")

if config.config_file_name is not None:
    fileConfig(config.config_file_name)


target_metadata = Base.metadata
url = config.get_main_option("sqlalchemy.url")


def truncate():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    
    with connectable.connect() as connection:
        Session = sessionmaker(bind=connection)
        session = Session() 
        session.execute(text('SET FOREIGN_KEY_CHECKS = 0;'))
        for table in target_metadata.sorted_tables:
            session.execute(table.delete())
        session.execute(text('SET FOREIGN_KEY_CHECKS = 1;'))
        session.commit()
        
if __name__=="__main__":
    truncate()

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment