utils.py 文件源码

def kill_db_processes(database_url, db_name=None, db_user_name=None):
    url = make_url(database_url)
    if url.get_backend_name() == 'sqlite':
        return [], []
    processes = get_db_processes(database_url)

    all_procs = []
    killed_procs = []
    with isolated_nullpool_engine(database_url) as engine:
        for process in processes:
            logger.debug(
                'process: Id:%s User:%s db:%s Command:%s State:%s Info:%s',
                process.Id, process.User, process.db, process.Command,
                process.State, process.Info)
            all_procs.append(process)
            if process.db == db_name and process.User == db_user_name:
                if process.Info != 'SHOW PROCESSLIST':
                    logger.debug('killing process %s on db %s owned by %s',
                                 process.Id, process.db, process.User)

                    engine.execute('KILL %s' % process.Id)
                    killed_procs.append(process)
        return all_procs, killed_procs