Python pymongo 模块，MongoClient() 实例源码

def _connect(self, address, lazy_connection=False):
        """Set up a connection to the MongoDB server.

        Parameters:
            address: MongoDB server address.
            lazy_connection: avoid testing if the connection is working while
                initializing it.
        """
        client = pymongo.MongoClient(address,
            serverSelectionTimeoutMS=FLAGS.mongodb_connection_timeout)
        if lazy_connection:
            return client

        # Send a query to the server to see if the connection is working.
        try:
            client.server_info()
        except pymongo.errors.ServerSelectionTimeoutError as e:
            logging.error("Unable to connect to %s.", address)
            client = None

        return client

def __loadTicksFromMongo(self,host,port,dbName,symbolName,startDatetimeStr,endDatetimeStr):
        """mid
        ??mongodb?????????????????
        """
        mongoConnection = MongoClient( host=host,port=port)
        collection = mongoConnection[dbName][symbolName]   

        startDate = dt.datetime.strptime(startDatetimeStr, '%Y-%m-%d %H:%M:%S')
        endDate = dt.datetime.strptime(endDatetimeStr, '%Y-%m-%d %H:%M:%S')  
        cx = collection.find({'datetime': {'$gte': startDate, '$lte': endDate}})    

        tickDatetimeNums = []
        tickPrices = []
        for d in cx:
            tickDatetimeNums.append(mpd.date2num(d['datetime']))
            tickPrices.append(d['lastPrice'])
        return tickDatetimeNums,tickPrices

    #----------------------------------------------------------------------

def validate_all_collections():
    """ Connecto to mongo and run db.collection.validate() on everything """
    retry_count = 0
    try:
        client = pymongo.MongoClient("localhost", 27017, maxPoolSize=50)
    except Exception as exc:
        if retry_count > 20:
            raise Exception("Retries exceeded") from exc
        retry_count += 1
        sleep(6)
    for db in (client[name] for name in
               client.database_names()
               if name != "local"):
        for collection in db.collection_names(include_system_collections=False):
            if db.validate_collection(collection, scandata=True, full=True)['errors']:
                raise ValidationFailed("Collection failed to validate", collection)

def process_spider_output(self, response, result, spider):
        """record this page
        """
        mongo_uri=spider.crawler.settings.get('MONGO_URI')
        mongo_db=spider.crawler.settings.get('MONGO_DB')
        client = pymongo.MongoClient(mongo_uri)
        db = client[mongo_db]
        def add_field(request, response):
            if isinstance(request, Request):
                db[self.collection_name].update_one(
                    {},
                    {'$set': {'page_url': response.request.url}},
                    upsert=True)
            return True
        ret = [req for req in result if add_field(req, response)]
        client.close()
        return ret

def validate_all_human_protein():
    # runs all proteins through the validator
    # and generates a log file

    coll = MongoClient().wikidata_src.mygene
    metadata_coll = MongoClient().wikidata_src.mygene_sources
    metadata = metadata_coll.find_one()
    doc_filter = {'taxid': 9606, 'entrezgene': {'$exists': True}}
    docs = coll.find(doc_filter)
    print("total number of records: {}".format(coll.find(doc_filter).count()))

    validate_type = 'eukaryotic'
    docs = HelperBot.validate_docs(docs, validate_type, 'P351')
    records = HelperBot.tag_mygene_docs(docs, metadata)

    _ = list(records)

def test_make_gene_class():
    coll = MongoClient().wikidata_src.mygene
    metadata_coll = MongoClient().wikidata_src.mygene_sources
    metadata = metadata_coll.find_one()
    doc_filter = {'_id': '100861512'}
    docs = coll.find(doc_filter)
    print("total number of records: {}".format(coll.find(doc_filter).count()))

    validate_type = 'eukaryotic'
    docs = HelperBot.validate_docs(docs, validate_type, 'P351')
    records = HelperBot.tag_mygene_docs(docs, metadata)
    record = next(records)

    organism_info = {
        "name": "Homo sapiens",
        "type": "mammalian",
        "wdid": "Q15978631",
        'taxid': 9606
    }

    login = wdi_login.WDLogin(WDUSER, WDPASS)

    gene = Gene(record, organism_info, login)
    gene.create_item(fast_run=False, write=True)
    gene.remove_deprecated_statements()

def validate_all_human_genes():
    # runs all genes through the validator
    # and generates a log file

    coll = MongoClient().wikidata_src.mygene
    metadata_coll = MongoClient().wikidata_src.mygene_sources
    metadata = metadata_coll.find_one()
    doc_filter = {'taxid': 9606, 'entrezgene': {'$exists': True}}
    docs = coll.find(doc_filter)
    print("total number of records: {}".format(coll.find(doc_filter).count()))

    validate_type = 'eukaryotic'
    docs = HelperBot.validate_docs(docs, validate_type, 'P351')
    records = HelperBot.tag_mygene_docs(docs, metadata)

    _ = list(records)

def connect(self):
        try:
            logging.debug("Getting MongoDB connection to %s (replicaSet=%s, readPreference=%s, readPreferenceTags=%s, ssl=%s)" % (
                self.uri,
                self.replset,
                self.read_pref,
                self.do_rp_tags,
                self.do_ssl(),
            ))
            conn = MongoClient(**self.client_opts())
            if self.do_connect:
                conn['admin'].command({"ping": 1})
        except (ConnectionFailure, OperationFailure, ServerSelectionTimeoutError), e:
            logging.error("Unable to connect to %s! Error: %s" % (self.uri, e))
            raise DBConnectionError(e)
        if conn is not None:
            self._conn = conn
        return self._conn

def mongo_no_context_get_job(job_id):
    """
    Get job object from MongoDB.
    This does not use context object from Flask.

    Parameters
    ----------
    job_id: str

    Returns
    -------
    dict
        Job object
    """
    client = MongoClient(MONGO_URI)
    db = client[MONGO_DBNAME]
    key = dict(_id=ObjectId(job_id))
    response = db.jobs.find_one(key)
    return response

def mongo_no_context_get_tasks(job_id):
    """
    Get all tasks for a job from MongoDB.
    This does not use context object from Flask.

    Parameters
    ----------
    job_id: str

    Returns
    -------
    list(dict)
        All task objects for given job
    """
    client = MongoClient(MONGO_URI)
    db = client[MONGO_DBNAME]
    key = dict(job_id=job_id)
    response = list(db.tasks.find(key))
    return response

def mongo_no_context_get_task(job_id, task_id):
    """
    Get a task from MongoDB.
    This does not use context object from Flask.

    Parameters
    ----------
    job_id: str
    task_id: int

    Returns
    -------
    dict
        task object
    """
    client = MongoClient(MONGO_URI)
    db = client[MONGO_DBNAME]
    key = dict(job_id=job_id, task_id=task_id)
    response = db.tasks.find_one(key)
    return response

def mongo_no_context_add_tasks(tasks):
    """
    Add tasks to MongoDB.
    This does not use context object from Flask.

    Parameters
    ----------
    tasks: list(dict)
        List of all task objects.

    Returns
    -------
    dict
        response from MongoDB.
    """
    client = MongoClient(MONGO_URI)
    db = client[MONGO_DBNAME]
    response = db.tasks.insert_many(tasks)
    return response

def init():

    connection = MongoClient(secret.mongo_url, secret.mongo_port)
    db = connection[secret.mongo_db]
    db.authenticate(secret.mongo_user, urllib.quote_plus(secret.mongo_pass))

    r = praw.Reddit(user_agent="Samachar Bot for /r/india by /u/sallurocks")
    scopes = {u'edit', u'submit', u'read', u'privatemessages', u'identity', u'history'}
    oauth_helper = PrawOAuth2Mini(r, app_key=secret.news_app_key,
                                  app_secret=secret.news_app_secret,
                                  access_token=secret.news_access_token,
                                  refresh_token=secret.news_refresh_token, scopes=scopes)

    init_object = {'db': db,
                   'reddit': r,
                   'oauth': oauth_helper,
                   'goose': Goose()}

    return init_object

def get_database(cred, **mongo_client_kwargs):
    """Connect to a database given a credential dict.

    Args:
        cred (dict): {database, [host, port, username, password]}

    Returns:
        pymongo.database.Database: The database object.
    """
    # respect potential multiprocessing fork
    mc_kwargs = dict(connect=False)
    mc_kwargs.update(mongo_client_kwargs)
    conn = MongoClient(
        cred.get('host', 'localhost'),
        cred.get('port', 27017),
        **mc_kwargs)
    db = conn[cred['database']]
    if cred.get('username'):
        db.authenticate(cred['username'], cred['password'])
    return db

def connect(self, url, max_retries, retry_interval):
        connection_options = pymongo.uri_parser.parse_uri(url)
        del connection_options['database']
        del connection_options['username']
        del connection_options['password']
        del connection_options['collection']
        pool_key = tuple(connection_options)

        if pool_key in self._pool:
            client = self._pool.get(pool_key)()
            if client:
                return client
        splitted_url = netutils.urlsplit(url)
        log_data = {'db': splitted_url.scheme,
                    'nodelist': connection_options['nodelist']}
        LOG.info('Connecting to %(db)s on %(nodelist)s' % log_data)
        try:
            client = MongoProxy(pymongo.MongoClient(url),
                                max_retries, retry_interval)
        except pymongo.errors.ConnectionFailure as e:
            LOG.warning(_('Unable to connect to the database server: '
                        '%(errmsg)s.') % {'errmsg': e})
            raise
        self._pool[pool_key] = weakref.ref(client)
        return client

def get_conn():
    """
    Get a database connection

    Ensures that only one global database connection exists per thread.
    If the connection does not exist a new one is created and returned.
    """

    if external_client is not None:
        return external_client

    global __client, __connection
    if not __connection:
        try:
            __client = MongoClient(mongo_addr, mongo_port)
            __connection = __client[mongo_db_name]
        except ConnectionFailure:
            raise SevereInternalException("Could not connect to mongo database {} at {}:{}".format(mongo_db_name, mongo_addr, mongo_port))
        except InvalidName as error:
            raise SevereInternalException("Database {} is invalid! - {}".format(mongo_db_name, error))

    return __connection

def save_to_database(list_id, scangroup_id):
    return db_connector.saveSingleUrl.s(list_id, scangroup_id)
    # state = db_connector.SaveScan(list_id, scangroup_id, urls)
    # # TODO The following is just error handling for the insert - will probably also have to be moved (statekeeping in MongoDB)
    # client = MongoClient(config.MONGODB_URL)
    # db = client['PrangerDB']
    # if state.startswith('error'):
    #     db.ScanGroup.update({'_id': ObjectId(scangroup_id)}, {'$set': {'state': "error during SaveScan - %s" % state}})
    #     print "error during SaveScan - %s" % state

    # elif state.startswith('success'):
    #     db.ScanGroup.update({'_id': ObjectId(scangroup_id)}, {'$set': {'state': 'finish'}})
    #     db.ScanGroup.update({'_id': ObjectId(scangroup_id)}, {'$set': {'progress': "finish"}})
    #     db.ScanGroup.update({'_id': ObjectId(scangroup_id)}, {'$set':{'progress_timestamp': datetime.now().isoformat()}}, upsert=False)

    # else:
    #     db.ScanGroup.update({'_id': ObjectId(scangroup_id)}, {'$set': {'state': 'unknown error during SaveScan: no status returned'}})
    #     print "unknown error during SaveScan: no status returned"

def __init__(self, database, collection, host=None):
        """

        :param host: ("localhost", 27017)
        :param database:
        :param collection:
        :return:
        """
        host = host or ("localhost", 27017)
        self.ip, self.port = host

        # ????
        self.client = pymongo.MongoClient(self.ip, self.port)
        self.log = self.client[database][collection]

        # ?????

        self.deals = None

def on_change():
    '''
    called when there is a change in the list of IPs and ports for this backend
    '''
    hostname = socket.gethostname()
    ip = get_ip()
    local_mongo = MongoClient(ip, connect=False)

    try:
        repl_status = local_mongo.admin.command('replSetGetStatus')
        is_mongo_primary = repl_status['myState'] == 1
        # ref https://docs.mongodb.com/manual/reference/replica-states/
    except Exception as e:
        log.error(e, 'unable to get primary status')
        return False

    if is_mongo_primary:
        return mongo_update_replset_config(local_mongo, ip)
    else:
        return True

# ---------------------------------------------------------

def get_collection(collection_name, custom_mongo_client=None):
    """
    Return the collection

    :type collection_name: str
    :param collection_name:
    :type custom_mongo_client: MongoClient
    :param custom_mongo_client:
    :rtype: Collection
    :return:
    """
    if custom_mongo_client is None:
        custom_mongo_client = get_mongo_client()

    db = custom_mongo_client[config.db_name]
    return db[collection_name]

def get(self):
        client = pymongo.MongoClient(config.MONGO_URI)
        db = client[config.MONGO_DATABASE]

        problems = db['problems'].find({'oj': 'poj'}, {'problem_id': 1, 'title': 1})

        problem_list = []
        problem_num = 0;
        for one in problems:
            problem = {
                'problem_id': one['problem_id'],
                'title': one['title']
            }

            problem_list.append(problem)
            problem_num += 1

        return {
            'problem_num': problem_num,
            'problem_list': problem_list
        }

def post(self, username):
        get_user = AccountCrawler()
        get_user.crawl('poj', username, request.json['password'])

        client = pymongo.MongoClient(config.MONGO_URI)
        db = client[config.MONGO_DATABASE]
        user_info = db['users'].find_one({'oj': 'poj', 'username': username})
        client.close()

        if user_info is None:
            return {
                'status': 404,
                'message': 'not found'
            }

        return {
            'username': user_info['username'],
            'status': 200,
            'submit': user_info['submit'],
            'oj': user_info['oj'],
            'accept': user_info['accept'],
            'rank': user_info['rank'],
            'solved': dict(user_info['solved'])
        }

def loadTick(self, dbName, collectionName, days):
        """???????Tick???startDate?datetime??"""
        startDate = datetime.now()

        d = {'datetime': {'$lte': startDate}}
        host, port, logging = loadMongoSetting()
        client = pymongo.MongoClient(host, port)
        collection = client[dbName][collectionName]

        cursor = collection.find(d).limit(days * 10 * 60 * 120)

        l = []
        if cursor:
            for d in cursor:
                tick = CtaTickData()
                tick.__dict__ = d
                l.append(tick)

        return l

        # ----------------------------------------------------------------------

    # ----------------------------------------------------------------------

def loadBar(self, dbName, collectionName, days):
        """???????Bar???startDate?datetime??"""
        startDate = datetime.now()

        d = {'datetime': {'$lte': startDate}}
        host, port, logging = loadMongoSetting()
        client = pymongo.MongoClient(host, port)
        collection = client[dbName][collectionName]

        cursor = collection.find(d).limit(days * 10 * 60)

        l = []
        if cursor:
            for d in cursor:
                bar = CtaBarData()
                bar.__dict__ = d
                l.append(bar)

        return l

        # ----------------------------------------------------------------------

    # ----------------------------------------------------------------------

def dbConnect(self):
        """??MongoDB???"""
        if not self.dbClient:
            # ??MongoDB???
            host, port, logging = loadMongoSetting()

            try:
                # ??MongoDB????????0.5?
                self.dbClient = MongoClient(host, port, connectTimeoutMS=500)

                # ??server_info?????????????????????
                self.dbClient.server_info()

                self.writeLog(u'MongoDB????')

                # ????????????????????
                if logging:
                    self.eventEngine.register(EVENT_LOG, self.dbLogging)

            except ConnectionFailure:
                self.writeLog(u'MongoDB????')

    #----------------------------------------------------------------------

def write_analytic(text, classname):
    time = datetime.now()
    analytic_id = classname
    # + str(time.year) + str(time.month) + str(time.day) + str(time.hour) + str(time.minute) + str(time.second)

    with open(ANALYTICS_OPALS + analytic_id + '.py', 'w') as alg:
        alg.write(text)

    #get the metadata from the file
    metadata = get_metadata(analytic_id)
    metadata['analytic_id'] = analytic_id

    client = pymongo.MongoClient(MONGO_HOST, MONGO_PORT)
    col = client[ANALYTICS_DB_NAME][ANALYTICS_COL_NAME]

    col.insert(metadata)

def connect_to_client(url="mongodb://localhost:27017/"):
    """
            Fill:

                Connect to a mongodb server accessible via the given url

            Args:

                url (str): url of the mongodb server

            Returns:

                mongodb client

    """
    return MongoClient(url)

def build_new_mongo_databases_and_collection(client):
    """
            Fill:

                Create the toplevel mongodb for TAXII, discovery_database, with its two collections:
                discovery_information and api_root_info

            Args:

                client (pymongo.MongoClient): mongodb client connection

            Returns:

                discovery_database object


    """
    db = client["discovery_database"]
    db["discovery_information"]
    db["api_root_info"]
    return db

def __init__(self):
        self.client = pymongo.MongoClient(
            settings['MONGO_HOST'],
            settings['MONGO_PORT']
        )
        self.db = self.client[settings['MONGO_DB']]
        self.collection = self.db[settings['MONGO_COLLECTION']]

def init_client():
    client = pymongo.MongoClient(config['db_host'], config['db_port'])
    if len(config['db_user']) != 0:
        admin = client[config['db_name']]
        admin.authenticate(config['db_user'], config['db_pass'])
    return client


# ??????tor??????http???

def setup_db():
    """ Creates a mongodb instance and shuts it down after testing has concluded. """

    client = MongoClient(api.config.testing_mongo_addr,
                         api.config.testing_mongo_port)[api.config.testing_mongo_db_name]

    if len(client.collection_names()) != 0:
        client.connection.drop_database(api.config.testing_mongo_db_name)

    #Set debug client for mongo
    if api.common.external_client is None:
        api.common.external_client = client

    return client

def teardown_db():
    """ Drops the db and shuts down the mongodb instance. """
    client = MongoClient(api.config.testing_mongo_addr,
                         api.config.testing_mongo_port)[api.config.testing_mongo_db_name]
    client.connection.drop_database(api.config.testing_mongo_db_name)
    client.connection.disconnect()

def get_conn():
    """
    Get a database connection

    Ensures that only one global database connection exists per thread.
    If the connection does not exist a new one is created and returned.
    """

    global __client, __connection
    if not __connection:
        try:
            # Allow more complex mongodb connections
            conf = api.app.app.config
            if conf["MONGO_USER"] and conf["MONGO_PW"]:
                uri = "mongodb://{}:{}@{}:{}/{}?authMechanism=SCRAM-SHA-1".format(
                        conf["MONGO_USER"],
                        conf["MONGO_PW"],
                        conf["MONGO_ADDR"],
                        conf["MONGO_PORT"],
                        conf["MONGO_DB_NAME"])
            else:
                uri = "mongodb://{}:{}/{}".format(
                        conf["MONGO_ADDR"],
                        conf["MONGO_PORT"],
                        conf["MONGO_DB_NAME"])

            __client = MongoClient(uri)
            __connection = __client[conf["MONGO_DB_NAME"]]
        except ConnectionFailure:
            raise SevereInternalException("Could not connect to mongo database {} at {}:{}".format(mongo_db_name, mongo_addr, mongo_port))
        except InvalidName as error:
            raise SevereInternalException("Database {} is invalid! - {}".format(mongo_db_name, error))

    return __connection

def connect(self):
        """
        Connect to the database and get a reference to the Mongo collection.

        :returns: the mongo collection.
        """
        self.conn = MongoClient(self.databaseUri)
        self.database = self.conn[self.databaseName]
        return self.database[self.collection]

def getTableList(uri, internalTables=False, **kwargs):
        """
        Get a list of known databases, each of which has a list of known
        collections from the database.  This is of the form [{'database':
        (database 1), 'tables': [{'table': (collection 1)}, {'table':
        (collection 2)}, ...]}, {'database': (database 2), 'tables': [...]},
        ...]

        :param uri: uri to connect to the database.
        :param internaltables: True to return tables about the database itself.
            Ignored for Mongo.
        :returns: A list of known collections.
        """
        conn = MongoClient(uri)
        databaseName = base.databaseFromUri(uri)
        if databaseName is None:
            databaseNames = conn.database_names()
        else:
            databaseNames = [databaseName]
        results = []
        for name in databaseNames:
            database = conn[name]
            results.append({
                'database': name,
                'tables': [{'table': collection, 'name': collection}
                           for collection in database.collection_names(False)]
            })
        return results

def Conn(self):
        self.client = pymongo.MongoClient(self.ip,self.port)
        self.connection=self.client.stock #storage stock information
        self.index=self.client.index #storage index
        self.pool=self.client.pool  #storate pool
        self.treasure=self.client.treasure
        #print self.connection.collection_names()
        #print self.index.collection_names()
        #print self.pool.collection_names()

def __init__(self):
        try:
            self.Client = pymongo.MongoClient(host=self.HOST, port=self.PORT)
            self.db = self.Client.yitu8
            assert self.db.authenticate(self.user, self.pwd), "mongo???????!"
        except Exception as err:
            logging.error("mongo connect error: {}".format(str(err)))

def get_connection(self):
        """
        Get the most secure kind of connection available.
        Returns:
          pymongo.MongoClient instance

        """
        fqdn, port = self.cred['nodelist'][0]
        if hasattr(self, 'conn'):
            self.conn.close()
            return self.get_plain_connection(fqdn, port)
        else:
            return self.get_tls_connection(fqdn, port)

def _connDB(self):
        configuration = MongoConfig()
        mongo_client = pymongo.MongoClient(host=configuration.host, port=configuration.port)
        self._client = mongo_client
        try:
            db = mongo_client[configuration.db_name]
            if configuration.username is not None:
                db.authenticate(configuration.username, password=configuration.password)
            self._currentDB = db
        except:
            pass

def __init__(self, *args, **kwargs):
        BaseCollection.__init__(self)
        if 'passed_mongo' in kwargs:
            self.mongo = kwargs['passed_mongo']
            self.mongo_database = self.mongo[args[2]]
            if args[0] and args[1]:
                self.mongo_database.authenticate(args[0], args[1])
            self.mongo_collection = self.mongo_database[args[3]]
        else:
            self.mongo = pymongo.MongoClient(args[0], int(args[1]))
            self.mongo_database = self.mongo[args[4]]
            if args[2] and args[3]:
                self.mongo_database.authenticate(args[2], args[3])
            self.mongo_collection = self.mongo_database[args[5]]

def test_pass_in_mongo(self):
        mongo_to_pass = pymongo.MongoClient(config['mongo']['host'], int(config['mongo']['port']))
        collection = MongoCollection(
            config['mongo']['user'],
            config['mongo']['password'],
            config['mongo']['database'],
            config['mongo']['collection'],
            passed_mongo=mongo_to_pass
        )
        self.assertTrue(len(list(collection.set_limit(10).get_iterator())) > 0)

def _get_connection(self, host, port):
        ckey = "{}:{}".format(host, port)
        conn = self._conns.get(ckey, None)
        if conn is None:
            mps = ('max_pool_size' if pymongo.version_tuple[0] == 2
                   else 'maxPoolSize')
            conn = pymongo.MongoClient(host, port, **{mps: self.MAX_POOL})
            self._conns[ckey] = conn
        return conn

def stream_events(self, inputs, ew):
        for input_name, input_item in inputs.inputs.iteritems():
            host = input_item["server"]
            port = input_item["port"]
            if not port is None:
                port = int(port)
            client = pymongo.MongoClient(host, port)
            self.stream_events_mongo(input_name, input_item, client, ew)

def QA_util_sql_mongo_setting(ip='127.0.0.1', port=27017):
    QA_sql_mongo_client = pymongo.MongoClient(ip, int(port))
    QA_util_log_info('ip:{},port:{}'.format(str(ip), str(port)))
    return QA_sql_mongo_client

# async

def get(self):
        client = MongoClient(DB_URI)
        database = client[DB_NAME]
        collection = database.nationalparks

        collection.remove({})
        collection.create_index([('Location', GEO2D)])

        with open(DATASET_FILE, 'r') as fp:
            entries = []

            for data in fp.readlines():
                entry = json.loads(data)

                loc = [entry['coordinates'][1], entry['coordinates'][0]]
                entry['Location'] = loc

                entries.append(entry)

                if len(entries) >= 1000:
                    collection.insert_many(entries)
                    entries = []

            if entries:
                collection.insert_many(entries)

        return 'Items inserted in database: %s' % collection.count()

def get(self):
        client = MongoClient(DB_URI)
        database = client[DB_NAME]
        collection = database.nationalparks

        return format_result(collection.find())

def get(self):
        args = request.args

        box = [[float(args['lon1']), float(args['lat1'])],
               [float(args['lon2']), float(args['lat2'])]]

        query = {'Location': {'$within': {'$box': box}}}

        client = MongoClient(DB_URI)
        database = client[DB_NAME]
        collection = database.nationalparks

        return format_result(collection.find(query))

def insertMongoDB(items):
    collection_name = 'wechat'
    client = pymongo.MongoClient(MONGO_URI)
    db = client[MONGO_DATABASE]
    for item in items:
        item['_id'] = str(ObjectId())
        db[collection_name].insert(dict(item))

def open_spider(self, spider):
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]

def test_bson(self):
        import pymongo
        outputs = run(
            self.analysis,
            inputs={
                'a': {
                    'format': 'objectlist.bson',
                    'mode': 'mongodb',
                    'db': 'test',
                    'collection': 'a'
                },
                'b': {
                    'format': 'objectlist.bson',
                    'mode': 'mongodb',
                    'db': 'test',
                    'collection': 'b'
                }
            },
            outputs={
                'c': {
                    'format': 'objectlist.bson',
                    'mode': 'mongodb',
                    'db': 'test',
                    'collection': 'temp'
                }
            })
        self.assertEqual(outputs['c']['format'], 'objectlist.bson')
        coll = pymongo.MongoClient('mongodb://localhost')['test']['temp']
        self.assertEqual([d for d in coll.find()], [self.aobj, self.bobj])