Function delete_from_all_tables

lib/sqlalchemy/testing/provision.py:572–601

an absolutely foolproof delete from all tables routine. dialects should override this to add special instructions like disable constraints etc.

(connection, cfg, metadata)

@register.init
def delete_from_all_tables(connection, cfg, metadata):
    """an absolutely foolproof delete from all tables routine.

    dialects should override this to add special instructions like
    disable constraints etc.

    """
    savepoints = getattr(cfg.requirements, "savepoints", False)
    if savepoints:
        savepoints = savepoints.enabled

    inspector = inspect(connection)

    for table in reversed(
        [
            t
            for (t, fks) in sort_tables_and_constraints(
                metadata.tables.values()
            )
            if t is not None
            # remember that inspector.get_table_names() is cached,
            # so this emits SQL once per unique schema name
            and t.name in inspector.get_table_names(schema=t.schema)
        ]
    ):
        if savepoints:
            with connection.begin_nested():
                connection.execute(table.delete())
        else:
            connection.execute(table.delete())


@register.init