Last active
April 5, 2023 03:50
-
-
Save absent1706/3ccc1722ea3ca23a5cf54821dbc813fb to your computer and use it in GitHub Desktop.
Sqlalchemy: Truncate all tables
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
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
It may be enough to disable a foreign key checks just for the current session:
con.execute('SET SESSION FOREIGN_KEY_CHECKS = ON')