scout/adapter/utils.py
def check_connection(host='localhost', port=27017, username=None, password=None,
                     authdb=None, max_delay=1):
    """Check if a connection could be made to the mongo process specified

    Args:
        host(str)
        port(int)
        username(str)
        password(str)
        authdb (str): database to to for authentication
        max_delay(int): Number of milliseconds to wait for connection

    Returns:
        bool: If connection could be established
    """
    #uri looks like:
    #mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]
    if username and password:
        uri = ("mongodb://{}:{}@{}:{}/{}"
               .format(quote_plus(username), quote_plus(password), host, port, authdb))
        log_uri = ("mongodb://{}:****@{}:{}/{}"
               .format(quote_plus(username), host, port, authdb))
    else:
        log_uri = uri = "mongodb://%s:%s" % (host, port)
    
    LOG.info("Test connection with uri: %s", log_uri)
    client = MongoClient(uri, serverSelectionTimeoutMS=max_delay)
    try:
        client.server_info()
    except (ServerSelectionTimeoutError,OperationFailure) as err:
        LOG.warning(err)
        return False

    return True
def check_connection(host='localhost', port=27017, username=None, password=None,
                     authdb=None, max_delay=1):
    """Check if a connection could be made to the mongo process specified

    Args:
        host(str)
        port(int)
        username(str)
        password(str)
        authdb (str): database to to for authentication
        max_delay(int): Number of milliseconds to wait for connection

    Returns:
        bool: If connection could be established
    """
    #uri looks like:
    #mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]
    if username and password:
        uri = ("mongodb://{}:{}@{}:{}/{}"
               .format(quote_plus(username), quote_plus(password), host, port, authdb))
        log_uri = ("mongodb://{}:****@{}:{}/{}"
               .format(quote_plus(username), host, port, authdb))
    else:
        log_uri = uri = "mongodb://%s:%s" % (host, port)
    
    LOG.info("Test connection with uri: %s", log_uri)
    client = MongoClient(uri, serverSelectionTimeoutMS=max_delay)
    try:
        client.server_info()
    except (ServerSelectionTimeoutError,OperationFailure) as err:
        LOG.warning(err)
        return False

    return True
Check if a connection could be made to the mongo process specified

    Args:
        host(str)
        port(int)
        username(str)
        password(str)
        authdb (str): database to to for authentication
        max_delay(int): Number of milliseconds to wait for connection

    Returns:
        bool: If connection could be established