def run_test(self):
        if not _HAVE_DNSPYTHON:
            raise unittest.SkipTest("DNS tests require the dnspython module")
        uri = test_case['uri']
        seeds = test_case['seeds']
        hosts = test_case['hosts']
        options = test_case.get('options')
        if seeds:
            seeds = split_hosts(','.join(seeds))
        if hosts:
            hosts = frozenset(split_hosts(','.join(hosts)))
        if options:
            for key, value in options.items():
                # Convert numbers / booleans to strings for comparison
                if isinstance(value, bool):
                    options[key] = 'true' if value else 'false'
                elif isinstance(value, (int, float)):
                    options[key] = str(value)

        if seeds:
            result = parse_uri(uri, validate=False)
            self.assertEqual(sorted(result['nodelist']), sorted(seeds))
            if options:
                opts = result['options']
                if 'readpreferencetags' in opts:
                    rpts = validate_read_preference_tags(
                        'readPreferenceTags', opts.pop('readpreferencetags'))
                    opts['readPreferenceTags'] = rpts
                self.assertEqual(result['options'], options)

            hostname = next(iter(client_context.client.nodes))[0]
            # The replica set members must be configured as 'localhost'.
            if hostname == 'localhost':
                copts = client_context.default_client_options.copy()
                if client_context.ssl is True:
                    # Our test certs don't support the SRV hosts used in these tests.
                    copts['ssl_match_hostname'] = False

                client = MongoClient(uri, **copts)
                # Force server selection
                client.admin.command('ismaster')
                wait_until(
                    lambda: hosts == client.nodes,
                    'match test hosts to client nodes')
        else:
            try:
                parse_uri(uri)
            except (ConfigurationError, ValueError):
                pass
            else:
                self.fail("failed to raise an exception")

 def test_split_hosts(self):
     self.assertRaises(ConfigurationError, split_hosts, "localhost:27017,")
     self.assertRaises(ConfigurationError, split_hosts, ",localhost:27017")
     self.assertRaises(ConfigurationError, split_hosts, "localhost:27017,,localhost:27018")
     self.assertEqual([("localhost", 27017), ("example.com", 27017)], split_hosts("localhost,example.com"))
     self.assertEqual(
         [("localhost", 27018), ("example.com", 27019)], split_hosts("localhost:27018,example.com:27019")
     )
     self.assertRaises(ConfigurationError, split_hosts, "::1", 27017)
     self.assertRaises(ConfigurationError, split_hosts, "[::1:27017")
     self.assertRaises(ConfigurationError, split_hosts, "::1")
     self.assertRaises(ConfigurationError, split_hosts, "::1]:27017")
     self.assertEqual([("::1", 27017)], split_hosts("[::1]:27017"))
     self.assertEqual([("::1", 27017)], split_hosts("[::1]"))

 def test_split_hosts(self):
     self.assertRaises(ConfigurationError, split_hosts,
                       'localhost:27017,')
     self.assertRaises(ConfigurationError, split_hosts,
                       ',localhost:27017')
     self.assertRaises(ConfigurationError, split_hosts,
                       'localhost:27017,,localhost:27018')
     self.assertEqual([('localhost', 27017), ('example.com', 27017)],
                      split_hosts('localhost,example.com'))
     self.assertEqual([('localhost', 27018), ('example.com', 27019)],
                      split_hosts('localhost:27018,example.com:27019'))
     self.assertRaises(ConfigurationError, split_hosts, '::1', 27017)
     self.assertRaises(ConfigurationError, split_hosts, '[::1:27017')
     self.assertRaises(ConfigurationError, split_hosts, '::1')
     self.assertRaises(ConfigurationError, split_hosts, '::1]:27017')
     self.assertEqual([('::1', 27017)], split_hosts('[::1]:27017'))
     self.assertEqual([('::1', 27017)], split_hosts('[::1]'))

 def test_split_hosts(self):
     self.assertRaises(ConfigurationError, split_hosts,
                       'localhost:27017,')
     self.assertRaises(ConfigurationError, split_hosts,
                       ',localhost:27017')
     self.assertRaises(ConfigurationError, split_hosts,
                       'localhost:27017,,localhost:27018')
     self.assertEqual([('localhost', 27017), ('example.com', 27017)],
                      split_hosts('localhost,example.com'))
     self.assertEqual([('localhost', 27018), ('example.com', 27019)],
                      split_hosts('localhost:27018,example.com:27019'))
     self.assertEqual([('/tmp/mongodb-27017.sock', None)],
                      split_hosts('/tmp/mongodb-27017.sock'))
     self.assertEqual([('/tmp/mongodb-27017.sock', None),
                       ('example.com', 27017)],
                      split_hosts('/tmp/mongodb-27017.sock,'
                                  'example.com:27017'))
     self.assertEqual([('example.com', 27017),
                       ('/tmp/mongodb-27017.sock', None)],
                      split_hosts('example.com:27017,'
                                  '/tmp/mongodb-27017.sock'))
     self.assertRaises(ValueError, split_hosts, '::1', 27017)
     self.assertRaises(ValueError, split_hosts, '[::1:27017')
     self.assertRaises(ValueError, split_hosts, '::1')
     self.assertRaises(ValueError, split_hosts, '::1]:27017')
     self.assertEqual([('::1', 27017)], split_hosts('[::1]:27017'))
     self.assertEqual([('::1', 27017)], split_hosts('[::1]'))

def main():
    module = AnsibleModule(
        argument_spec={
            'login_user':     {'aliases': ['user']},
            'login_password': {'aliases': ['password'], no_log: True},
            'login_host':     {'aliases': ['host'], 'default': 'localhost'},
            'login_port':     {'aliases': ['port'], 'default': 27017},
            'hosts':          {'aliases': ['members'], 'required': True},
            'replica_set':    {'aliases': ['replset'], 'required': True}
        },
        required_together=[['login_host', 'login_port']]
    )

    if not pymongo_found:
        module.fail_json(msg='Python module "pymongo" must be installed.')

    user, password, host, port, hosts, replset = (
        module.params[k] for k in ['login_user', 'login_password', 'login_host',
                                   'login_port', 'hosts', 'replica_set'])
    nodes = split_hosts(hosts)

    if not user and not password:
        user, password = read_mongocnf_creds()

    initiated = False
    try:
        try:
            client = MongoReplicaSetClient(hosts, replicaSet=replset,
                                           read_preference=ReadPreference.PRIMARY)
            initiated = True
        except ConfigurationError, e:
            if 'is not a member of replica set' in e.message:
                client = MongoClient(host, int(port), read_preference=ReadPreference.SECONDARY)
            else:
                module.fail_json(msg="Unable to connect: %s" % e)

        if user and password:
            try:
                client.admin.authenticate(user, password)
            except OperationFailure, e:
                pass  # try to continue, maybe admin account is not set yet

#.........这里部分代码省略.........

        .. versionchanged:: 2.5
           Added additional ssl options
        .. versionadded:: 2.4
        """
        if host is None:
            host = self.HOST
        if isinstance(host, basestring):
            host = [host]
        if port is None:
            port = self.PORT
        if not isinstance(port, int):
            raise TypeError("port must be an instance of int")

        seeds = set()
        username = None
        password = None
        db_name = None
        opts = {}
        for entity in host:
            if "://" in entity:
                if entity.startswith("mongodb://"):
                    res = uri_parser.parse_uri(entity, port)
                    seeds.update(res["nodelist"])
                    username = res["username"] or username
                    password = res["password"] or password
                    db_name = res["database"] or db_name
                    opts = res["options"]
                else:
                    idx = entity.find("://")
                    raise InvalidURI("Invalid URI scheme: "
                                     "%s" % (entity[:idx],))
            else:
                seeds.update(uri_parser.split_hosts(entity, port))
        if not seeds:
            raise ConfigurationError("need to specify at least one host")

        self.__nodes = seeds
        self.__host = None
        self.__port = None
        self.__is_primary = False
        self.__is_mongos = False

        # _pool_class option is for deep customization of PyMongo, e.g. Motor.
        # SHOULD NOT BE USED BY DEVELOPERS EXTERNAL TO 10GEN.
        pool_class = kwargs.pop('_pool_class', pool.Pool)

        options = {}
        for option, value in kwargs.iteritems():
            option, value = common.validate(option, value)
            options[option] = value
        options.update(opts)

        self.__max_pool_size = common.validate_positive_integer(
                                                'max_pool_size', max_pool_size)

        self.__cursor_manager = CursorManager(self)

        self.__repl = options.get('replicaset')
        if len(seeds) == 1 and not self.__repl:
            self.__direct = True
        else:
            self.__direct = False
            self.__nodes = set()

        self.__net_timeout = options.get('sockettimeoutms')

#.........这里部分代码省略.........

               client['__my_database__']

           Not::

               client.__my_database__
        """
        if host is None:
            host = self.HOST
        if isinstance(host, string_type):
            host = [host]
        if port is None:
            port = self.PORT
        if not isinstance(port, int):
            raise TypeError("port must be an instance of int")

        seeds = set()
        username = None
        password = None
        dbase = None
        opts = {}
        for entity in host:
            if "://" in entity:
                if entity.startswith("mongodb://"):
                    res = uri_parser.parse_uri(entity, port, False)
                    seeds.update(res["nodelist"])
                    username = res["username"] or username
                    password = res["password"] or password
                    dbase = res["database"] or dbase
                    opts = res["options"]
                else:
                    idx = entity.find("://")
                    raise InvalidURI("Invalid URI scheme: "
                                     "%s" % (entity[:idx],))
            else:
                seeds.update(uri_parser.split_hosts(entity, port))
        if not seeds:
            raise ConfigurationError("need to specify at least one host")

        # _pool_class, _monitor_class, and _condition_class are for deep
        # customization of PyMongo, e.g. Motor.
        pool_class = kwargs.pop('_pool_class', None)
        monitor_class = kwargs.pop('_monitor_class', None)
        condition_class = kwargs.pop('_condition_class', None)

        kwargs['document_class'] = document_class
        kwargs['tz_aware'] = tz_aware
        opts.update(kwargs)
        self.__options = options = ClientOptions(
            username, password, dbase, opts)

        self.__default_database_name = dbase
        self.__lock = threading.Lock()
        self.__cursor_manager = CursorManager(self)
        self.__kill_cursors_queue = []

        # Cache of existing indexes used by ensure_index ops.
        self.__index_cache = {}

        super(MongoClient, self).__init__(options.codec_options,
                                          options.read_preference,
                                          options.write_concern)

        self.__all_credentials = {}
        creds = options.credentials
        if creds:
            self._cache_credentials(creds.source, creds)

        self._topology_settings = TopologySettings(
            seeds=seeds,
            replica_set_name=options.replica_set_name,
            pool_class=pool_class,
            pool_options=options.pool_options,
            monitor_class=monitor_class,
            condition_class=condition_class,
            local_threshold_ms=options.local_threshold_ms,
            server_selection_timeout=options.server_selection_timeout)

        self._topology = Topology(self._topology_settings)
        if connect:
            self._topology.open()

        def target():
            client = self_ref()
            if client is None:
                return False  # Stop the executor.
            MongoClient._process_kill_cursors_queue(client)
            return True

        executor = periodic_executor.PeriodicExecutor(
            condition_class=self._topology_settings.condition_class,
            interval=common.KILL_CURSOR_FREQUENCY,
            min_interval=0,
            target=target)

        # We strongly reference the executor and it weakly references us via
        # this closure. When the client is freed, stop the executor soon.
        self_ref = weakref.ref(self, executor.close)
        self._kill_cursors_executor = executor
        executor.open()

    def __init__(self, hosts_or_uri=None, max_pool_size=10,
                 document_class=dict, tz_aware=False, **kwargs):
        """Create a new connection to a MongoDB replica set.

        The resultant connection object has connection-pooling built
        in. It also performs auto-reconnection when necessary. If an
        operation fails because of a connection error,
        :class:`~pymongo.errors.ConnectionFailure` is raised. If
        auto-reconnection will be performed,
        :class:`~pymongo.errors.AutoReconnect` will be
        raised. Application code should handle this exception
        (recognizing that the operation failed) and then continue to
        execute.

        Raises :class:`~pymongo.errors.ConnectionFailure` if
        the connection cannot be made.

        The `hosts_or_uri` parameter can be a full `mongodb URI
        <http://dochub.mongodb.org/core/connections>`_, in addition to
        a string of `host:port` pairs (e.g. 'host1:port1,host2:port2').
        If `hosts_or_uri` is None 'localhost:27017' will be used.

        :Parameters:
          - `hosts_or_uri` (optional): A MongoDB URI or string of `host:port`
            pairs. If a host is an IPv6 literal it must be enclosed in '[' and
            ']' characters following the RFC2732 URL syntax (e.g. '[::1]' for
            localhost)
          - `max_pool_size` (optional): The maximum size limit for
            each connection pool.
          - `document_class` (optional): default class to use for
            documents returned from queries on this connection
          - `tz_aware` (optional): if ``True``,
            :class:`~datetime.datetime` instances returned as values
            in a document by this :class:`Connection` will be timezone
            aware (otherwise they will be naive)
          - `replicaSet`: (required) The name of the replica set to connect to.
            The driver will verify that each host it connects to is a member of
            this replica set. Can be passed as a keyword argument or as a
            MongoDB URI option.

          Other optional parameters can be passed as keyword arguments:

          - `safe`: Use getlasterror for each write operation?
          - `j` or `journal`: Block until write operations have been commited
            to the journal. Ignored if the server is running without
            journaling. Implies safe=True.
          - `w`: (integer or string) If this is a replica set write operations
            won't return until they have been replicated to the specified
            number or tagged set of servers.
            Implies safe=True.
          - `wtimeoutMS`: Used in conjunction with `j` and/or `w`. Wait this
            many milliseconds for journal acknowledgement and/or write
            replication. Implies safe=True.
          - `fsync`: Force the database to fsync all files before returning
            When used with `j` the server awaits the next group commit before
            returning. Implies safe=True.
          - `socketTimeoutMS`: How long a send or receive on a socket can take
            before timing out.
          - `connectTimeoutMS`: How long a connection can take to be opened
            before timing out.
          - `ssl`: If True, create the connection to the servers using SSL.
          - `read_preference`: The read preference for this connection.
            See :class:`~pymongo.ReadPreference` for available options.
          - `slave_okay` or `slaveOk` (deprecated): Use `read_preference`
            instead.

        .. versionadded:: 2.1
        """
        self.__max_pool_size = max_pool_size
        self.__document_class = document_class
        self.__tz_aware = tz_aware
        self.__opts = {}
        self.__seeds = set()
        self.__hosts = None
        self.__arbiters = set()
        self.__writer = None
        self.__readers = []
        self.__pools = {}
        self.__index_cache = {}
        self.__auth_credentials = {}
        username = None
        db_name = None
        if hosts_or_uri is None:
            self.__seeds.add(('localhost', 27017))
        elif '://' in hosts_or_uri:
            res = uri_parser.parse_uri(hosts_or_uri)
            self.__seeds.update(res['nodelist'])
            username = res['username']
            password = res['password']
            db_name = res['database']
            self.__opts = res['options']
        else:
            self.__seeds.update(uri_parser.split_hosts(hosts_or_uri))

        for option, value in kwargs.iteritems():
            option, value = common.validate(option, value)
            self.__opts[option] = value

        self.__name = self.__opts.get('replicaset')
        if not self.__name:
#.........这里部分代码省略.........

#.........这里部分代码省略.........
        .. versionadded:: 1.1
           The `network_timeout` parameter.

        .. mongodoc:: connections
        """
        if host is None:
            host = self.HOST
        if isinstance(host, basestring):
            host = [host]
        if port is None:
            port = self.PORT
        if not isinstance(port, int):
            raise TypeError("port must be an instance of int")

        seeds = set()
        username = None
        password = None
        db = None
        options = {}
        for entity in host:
            if "://" in entity:
                if entity.startswith("mongodb://"):
                    res = uri_parser.parse_uri(entity, port)
                    seeds.update(res["nodelist"])
                    username = res["username"] or username
                    password = res["password"] or password
                    db = res["database"] or db
                    options = res["options"]
                else:
                    idx = entity.find("://")
                    raise InvalidURI("Invalid URI scheme: "
                                     "%s" % (entity[:idx],))
            else:
                seeds.update(uri_parser.split_hosts(entity, port))
        if not seeds:
            raise ConfigurationError("need to specify at least one host")

        self.__nodes = seeds
        self.__host = None
        self.__port = None
        self.__is_primary = False
        self.__is_mongos = False

        for option, value in kwargs.iteritems():
            option, value = common.validate(option, value)
            options[option] = value

        self.__max_pool_size = common.validate_positive_integer(
                                                'max_pool_size', max_pool_size)

        self.__cursor_manager = CursorManager(self)

        self.__repl = options.get('replicaset')
        if len(seeds) == 1 and not self.__repl:
            self.__direct = True
        else:
            self.__direct = False
            self.__nodes = set()

        if network_timeout is not None:
            if (not isinstance(network_timeout, (int, float)) or
                network_timeout <= 0):
                raise ConfigurationError("network_timeout must "
                                         "be a positive integer")
        self.__net_timeout = (network_timeout or
                              options.get('sockettimeoutms'))

#.........这里部分代码省略.........
            Implies safe=True.
          - `replicaset`: The name of the replica set to connect to. The driver
            will verify that the replica set it connects to matches this name.
            Implies that the hosts specified are a seed list and the driver should
            attempt to find all members of the set.

        .. seealso:: :meth:`end_request`
        .. versionchanged:: 2.0.1+
           Support `w` = integer or string.
        .. versionchanged:: 2.0
           `slave_okay` is a pure keyword argument. Added support for safe,
           and getlasterror options as keyword arguments.
        .. versionchanged:: 1.11
           Added `max_pool_size`. Completely removed previously deprecated
           `pool_size`, `auto_start_request` and `timeout` parameters.
        .. versionchanged:: 1.8
           The `host` parameter can now be a full `mongodb URI
           <http://dochub.mongodb.org/core/connections>`_, in addition
           to a simple hostname. It can also be a list of hostnames or
           URIs.
        .. versionadded:: 1.8
           The `tz_aware` parameter.
        .. versionadded:: 1.7
           The `document_class` parameter.
        .. versionchanged:: 1.4
           DEPRECATED The `pool_size`, `auto_start_request`, and `timeout`
           parameters.
        .. versionadded:: 1.1
           The `network_timeout` parameter.

        .. mongodoc:: connections
        """
        super(Connection, self).__init__(**kwargs)

        if host is None:
            host = self.HOST
        if isinstance(host, basestring):
            host = [host]
        if port is None:
            port = self.PORT
        if not isinstance(port, int):
            raise TypeError("port must be an instance of int")

        nodes = set()
        username = None
        password = None
        db = None
        options = {}
        for entity in host:
            if "://" in entity:
                if entity.startswith("mongodb://"):
                    res = uri_parser.parse_uri(entity, port)
                    nodes.update(res["nodelist"])
                    username = res["username"] or username
                    password = res["password"] or password
                    db = res["database"] or db
                    options = res["options"]
                else:
                    idx = entity.find("://")
                    raise InvalidURI("Invalid URI scheme: " "%s" % (entity[:idx],))
            else:
                nodes.update(uri_parser.split_hosts(entity, port))
        if not nodes:
            raise ConfigurationError("need to specify at least one host")
        self.__nodes = nodes

        self.__host = None
        self.__port = None

        if options:
            super(Connection, self)._set_options(**options)

        assert isinstance(max_pool_size, int), "max_pool_size must be an int"
        self.__max_pool_size = options.get("maxpoolsize") or max_pool_size
        if self.__max_pool_size < 0:
            raise ValueError("the maximum pool size must be >= 0")

        self.__cursor_manager = CursorManager(self)

        self.__repl = options.get("replicaset", kwargs.get("replicaset"))
        self.__network_timeout = network_timeout
        self.__pool = _Pool(self.__max_pool_size, self.__network_timeout)
        self.__last_checkout = time.time()

        self.__document_class = document_class
        self.__tz_aware = tz_aware

        # cache of existing indexes used by ensure_index ops
        self.__index_cache = {}
        self.__auth_credentials = {}

        if _connect:
            self.__find_node()

        if db and username is None:
            warnings.warn("must provide a username and password " "to authenticate to %s" % (db,))
        if username:
            db = db or "admin"
            if not self[db].authenticate(username, password):
                raise ConfigurationError("authentication failed")

#.........这里部分代码省略.........
           Added `auto_start_request` option.
           Added support for `host`, `port`, and `network_timeout` keyword
           arguments for compatibility with connection.Connection.
        .. versionadded:: 2.1
        """
        self.__opts = {}
        self.__seeds = set()
        self.__hosts = None
        self.__arbiters = set()
        self.__writer = None
        self.__readers = []
        self.__pools = {}
        self.__index_cache = {}
        self.__auth_credentials = {}

        self.__max_pool_size = common.validate_positive_integer(
                                        'max_pool_size', max_pool_size)
        self.__tz_aware = common.validate_boolean('tz_aware', tz_aware)
        self.__document_class = document_class

        # Compatibility with connection.Connection
        host = kwargs.pop('host', hosts_or_uri)

        port = kwargs.pop('port', 27017)
        if not isinstance(port, int):
            raise TypeError("port must be an instance of int")

        network_timeout = kwargs.pop('network_timeout', None)
        if network_timeout is not None:
            if (not isinstance(network_timeout, (int, float)) or
                network_timeout <= 0):
                raise ConfigurationError("network_timeout must "
                                         "be a positive integer")

        username = None
        db_name = None
        if host is None:
            self.__seeds.add(('localhost', port))
        elif '://' in host:
            res = uri_parser.parse_uri(host, port)
            self.__seeds.update(res['nodelist'])
            username = res['username']
            password = res['password']
            db_name = res['database']
            self.__opts = res['options']
        else:
            self.__seeds.update(uri_parser.split_hosts(host, port))

        for option, value in kwargs.iteritems():
            option, value = common.validate(option, value)
            self.__opts[option] = value

        if self.__opts.get('use_greenlets', False):
            if not have_gevent:
                raise ConfigurationError(
                    "The gevent module is not available. "
                    "Install the gevent package from PyPI."
                )
            self.pool_class = pool.GreenletPool
        else:
            self.pool_class = pool.Pool

        self.__auto_start_request = self.__opts.get('auto_start_request', True)
        self.__in_request = self.__auto_start_request
        self.__name = self.__opts.get('replicaset')
        if not self.__name:
            raise ConfigurationError("the replicaSet "
                                     "keyword parameter is required.")


        self.__net_timeout = (network_timeout or
                              self.__opts.get('sockettimeoutms'))
        self.__conn_timeout = self.__opts.get('connecttimeoutms')
        self.__use_ssl = self.__opts.get('ssl', False)
        if self.__use_ssl and not pool.have_ssl:
            raise ConfigurationError("The ssl module is not available. If you "
                                     "are using a python version previous to "
                                     "2.6 you must install the ssl package "
                                     "from PyPI.")

        super(ReplicaSetConnection, self).__init__(**self.__opts)
        if self.slave_okay:
            warnings.warn("slave_okay is deprecated. Please "
                          "use read_preference instead.", DeprecationWarning)

        self.refresh()

        if self.__opts.get('use_greenlets', False):
            monitor = MonitorGreenlet(self)
        else:
            monitor = MonitorThread(self)
        monitor.start()

        if db_name and username is None:
            warnings.warn("must provide a username and password "
                          "to authenticate to %s" % (db_name,))
        if username:
            db_name = db_name or 'admin'
            if not self[db_name].authenticate(username, password):
                raise ConfigurationError("authentication failed")

    def __init__(self, hosts_or_uri, max_pool_size=10,
                 document_class=dict, tz_aware=False, **kwargs):
        """Create a new connection to a MongoDB replica set.

        The resultant connection object has connection-pooling built
        in. It also performs auto-reconnection when necessary. If an
        operation fails because of a connection error,
        :class:`~pymongo.errors.ConnectionFailure` is raised. If
        auto-reconnection will be performed,
        :class:`~pymongo.errors.AutoReconnect` will be
        raised. Application code should handle this exception
        (recognizing that the operation failed) and then continue to
        execute.

        Raises :class:`~pymongo.errors.ConnectionFailure` if
        the connection cannot be made.

        The `hosts_or_uri` parameter can be a full `mongodb URI
        <http://dochub.mongodb.org/core/connections>`_, in addition to
        a string of `host:port` pairs (e.g. 'host1:port1,host2:port2').

        :Parameters:
          - `hosts_or_uri`: A MongoDB URI or string of `host:port` pairs.
            If a host is an IPv6 literal it must be enclosed in '[' and ']'
            characters following the RFC2732 URL syntax (e.g. '[::1]' for
            localhost)
          - `max_pool_size` (optional): The maximum size limit for
            each connection pool.
          - `document_class` (optional): default class to use for
            documents returned from queries on this connection
          - `tz_aware` (optional): if ``True``,
            :class:`~datetime.datetime` instances returned as values
            in a document by this :class:`Connection` will be timezone
            aware (otherwise they will be naive)
          - `replicaset`: (required) The name of the replica set to connect to.
            The driver will verify that each host it connects to is a member
            of this replica set.

          Other optional parameters can be passed as keyword arguments:

          - `safe`: Use getlasterror for each write operation?
          - `j` or `journal`: Block until write operations have been commited
            to the journal. Ignored if the server is running without
            journaling. Implies safe=True.
          - `w`: (integer or string) If this is a replica set write operations
            won't return until they have been replicated to the specified
            number or tagged set of servers.
            Implies safe=True.
          - `wtimeout`: Used in conjunction with `j` and/or `w`. Wait this many
            milliseconds for journal acknowledgement and/or write replication.
            Implies safe=True.
          - `fsync`: Force the database to fsync all files before returning
            When used with `j` the server awaits the next group commit before
            returning. Implies safe=True.
          - `socketTimeoutMS`: How long a send or receive on a socket can take
            before timing out.
          - `connectTimeoutMS`: How long a connection can take to be opened
            before timing out.

        .. versionadded:: 2.0.1+
        """
        warnings.warn("ReplicaSetConnection IS NOT READY "
                      "FOR PRODUCTION USE. USE AT YOUR OWN RISK!")

        self.__max_pool_size = max_pool_size
        self.__document_class = document_class
        self.__tz_aware = tz_aware
        self.__opts = {}
        self.__hosts = set()
        self.__arbiters = set()
        self.__writer = None
        self.__readers = []
        self.__reader_pools = {}
        self.__index_cache = {}
        self.__auth_credentials = {}
        self.__read_pref = SECONDARY
        username = None
        db_name = None
        if '://' in hosts_or_uri:
            res = uri_parser.parse_uri(hosts_or_uri)
            self.__seeds = res['nodes']
            username = res['username']
            password = res['password']
            db_name = res['database']
            self.__opts = res['options']
        else:
            self.__seeds = uri_parser.split_hosts(hosts_or_uri)

        for option, value in kwargs.iteritems():
            option, value = common.validate(option, value)
            self.__opts[option] = value

        self.__name = self.__opts.get('replicaset')
        if not self.__name:
            raise ConfigurationError("You must provide a replica set name.")
        self.__net_timeout = self.__opts.get('sockettimeoutms')
        self.__conn_timeout = self.__opts.get('connecttimeoutms')

        super(ReplicaSetConnection, self).__init__(**self.__opts)

#.........这里部分代码省略.........

    def __init__(self, 
                host=None, 
                port=27017, 
                mincached=0, 
                maxcached=0, 
                maxconnections=0, 
                maxusage=0, 
                dbname=None, 
                slave_okay=False, 
                *args, **kwargs):

        if isinstance(host, basestring):
            host = [host]
        else:
            assert isinstance(host, list)
        assert isinstance(port, int)
        assert isinstance(mincached, int)
        assert isinstance(maxcached, int)
        assert isinstance(maxconnections, int)
        assert isinstance(maxusage, int)
        assert isinstance(dbname, (str, unicode, None.__class__))
        assert isinstance(slave_okay, bool)

        if mincached and maxcached:
            assert mincached <= maxcached
        if maxconnections:
            assert maxconnections >= maxcached
            assert maxconnections >= mincached
        self._args, self._kwargs = args, kwargs
        self._maxusage = maxusage
        self._mincached = mincached
        self._maxcached = maxcached
        self._maxconnections = maxconnections
        self._idle_cache = [] # the actual connections that can be used
        self._condition = Condition()
        self._kwargs['slave_okay'] = self._slave_okay = slave_okay
        self._connections = 0

        nodes = set()
        username = None  # TODO: username/password ignored for now
        password = None
        for entity in host:
            if "://" in entity:
                if entity.startswith("mongodb://"):
                    res = uri_parser.parse_uri(entity, port)
                    nodes.update(res["nodelist"])
                    username = res["username"] or username
                    password = res["password"] or password
                    dbname = res["database"] or dbname
                else:
                    idx = entity.find("://")
                    raise ProgrammingError("Invalid URI scheme: "
                                     "%s" % (entity[:idx],))
            else:
                nodes.update(uri_parser.split_hosts(entity, port))
        if not nodes:
            raise ProgrammingError("Need to specify at least one host")
        self._nodes = nodes
        self._dbname = dbname

        # Establish an initial number of idle database connections:
        idle = [self.connection() for i in range(mincached)]
        while idle:
            self.cache(idle.pop())

#.........这里部分代码省略.........
            thread's lifetime.  This ensures consistent reads, even if you
            read after an unacknowledged write. Defaults to ``False``
          - `use_greenlets`: If ``True``, :meth:`start_request()` will ensure
            that the current greenlet uses the same socket for all
            operations until :meth:`end_request()`

        .. seealso:: :meth:`end_request`

        .. mongodoc:: connections

        .. versionadded:: 2.4
        """
        if host is None:
            host = self.HOST
        if isinstance(host, basestring):
            host = [host]
        if port is None:
            port = self.PORT
        if not isinstance(port, int):
            raise TypeError("port must be an instance of int")

        seeds = set()
        username = None
        password = None
        db = None
        opts = {}
        for entity in host:
            if "://" in entity:
                if entity.startswith("mongodb://"):
                    res = uri_parser.parse_uri(entity, port)
                    seeds.update(res["nodelist"])
                    username = res["username"] or username
                    password = res["password"] or password
                    db = res["database"] or db
                    opts = res["options"]
                else:
                    idx = entity.find("://")
                    raise InvalidURI("Invalid URI scheme: " "%s" % (entity[:idx],))
            else:
                seeds.update(uri_parser.split_hosts(entity, port))
        if not seeds:
            raise ConfigurationError("need to specify at least one host")

        self.__nodes = seeds
        self.__host = None
        self.__port = None
        self.__is_primary = False
        self.__is_mongos = False

        options = {}
        for option, value in kwargs.iteritems():
            option, value = common.validate(option, value)
            options[option] = value
        options.update(opts)

        self.__max_pool_size = common.validate_positive_integer("max_pool_size", max_pool_size)

        self.__cursor_manager = CursorManager(self)

        self.__repl = options.get("replicaset")
        if len(seeds) == 1 and not self.__repl:
            self.__direct = True
        else:
            self.__direct = False
            self.__nodes = set()

        self.__net_timeout = options.get("sockettimeoutms")
        self.__conn_timeout = options.get("connecttimeoutms")
        self.__use_ssl = options.get("ssl", False)
        if self.__use_ssl and not pool.have_ssl:
            raise ConfigurationError(
                "The ssl module is not available. If you "
                "are using a python version previous to "
                "2.6 you must install the ssl package "
                "from PyPI."
            )

        use_greenlets = options.get("use_greenlets", False)
        self.__pool = pool.Pool(
            None, self.__max_pool_size, self.__net_timeout, self.__conn_timeout, self.__use_ssl, use_greenlets
        )

        self.__document_class = document_class
        self.__tz_aware = common.validate_boolean("tz_aware", tz_aware)
        self.__auto_start_request = options.get("auto_start_request", False)

        # cache of existing indexes used by ensure_index ops
        self.__index_cache = {}
        self.__auth_credentials = {}

        super(MongoClient, self).__init__(**options)
        if self.slave_okay:
            warnings.warn("slave_okay is deprecated. Please " "use read_preference instead.", DeprecationWarning)

        if _connect:
            try:
                self.__find_node(seeds)
            except AutoReconnect, e:
                # ConnectionFailure makes more sense here than AutoReconnect
                raise ConnectionFailure(str(e))