(db_name, db_user, db_port, db_host, db_passwd)
HARD Truncate all DB Tables
HARD Truncate all DB Tables
HARD Truncate all DB Tables
def flush_db(db_name, db_user, db_port, db_host, db_passwd):
    """HARD Truncate all DB Tables"""
    db_host = db_host if db_host is not None else 'localhost'
    db_port = db_port if db_port is not None else 5432
    conn = get_db_conn(db_name, db_user, db_port, db_host, db_passwd)
    curs = conn.cursor()

    try:
        sql_dump = f"""SELECT tablename from pg_tables where tableowner = '{db_user}'"""
        curs.execute(sql_dump)
        pg_tables = curs.fetchall()
        for table in pg_tables:
            if table[0] == 'br_restoredbackup':
                continue
            print(f"Flushing Data : {table[0]}")
            curs.execute(f"TRUNCATE {table[0]} CASCADE;")
        conn.commit()
    except Exception:
        try:
            conn.rollback()
        except Exception:
            pass
        traceback.print_exc()
    finally:
        curs.close()
        conn.close()
def flush_db ( db_name , db_user , db_port , db_host , db_passwd ) : db_host = db_host if db_host is not None else 'localhost' db_port = db_port if db_port is not None else 5432 conn = get_db_conn ( db_name , db_user , db_port , db_host , db_passwd ) curs = conn . cursor ( ) try : sql_dump = f"""SELECT tablename from pg_tables where tableowner = '{db_user}'""" curs . execute ( sql_dump ) pg_tables = curs . fetchall ( ) for table in pg_tables : if table [ 0 ] == 'br_restoredbackup' : continue print ( f"Flushing Data : {table[0]}" ) curs . execute ( f"TRUNCATE {table[0]} CASCADE;" ) conn . commit ( ) except Exception : try : conn . rollback ( ) except Exception : pass traceback . print_exc ( ) finally : curs . close ( ) conn . close ( )