db_utils.py 文件源码

def get_dir_db_connection(*, dict_cursor: bool = False):
    try:
        connection = psycopg2.connect(
            dbname=DIR_DB_NAME, user=DIR_DB_USER, password=DIR_DB_PASSWORD,
            host=DIR_DB_HOST, port=DIR_DB_PORT)
    except psycopg2.OperationalError as e:
        logging.error('Unable to connect to Directory DB!\n%s', e)
        raise
    else:
        logging.debug('Connected to Directory DB: %s!', DIR_DB_NAME)
    if dict_cursor:
        cursor = connection.cursor(cursor_factory=RealDictCursor)
    else:
        cursor = connection.cursor()
    return connection, cursor