python pymongo使用mongodb入门

亦凉 2021-11-09 04:42 383阅读 0赞

python 使用mongodb是通过pymongo进行操作的

##### 安装： #####

pip install pymongo

##### 连接 #####

client = MongoClient('mongodb://localhost:27017/')

##### 创建db或连接db #####

client.test_database
    client['test-database']
    db = client.get_database("db_name")

认证：

db.authenticate('username', 'password')

##### 创建集合collection 或获取集合 #####

db.test_collection
    db['test-collection']

##### 查看集合 #####

db.list_collection_names()

##### 新增 #####

init_data = {"author": "tom", "text": "My first blog post!",
                 "tags": ["mongodb", "python", "pymongo"],
                 "date": datetime.datetime.utcnow()}
    
    post_id = collection.insert_one(init_data).inserted_id
    
    print post_id

##### 查询单文档 #####

可以根据条件进行查询， 也可以使用projectid进行查询，注意： project不是简单的str字符串

print collection.find_one({"_id": post_id})
    
    print collection.find_one({"author": "Mike"})

使用projectid查询

from bson.objectid import ObjectId
    
    
    def get(post_id):
        # Convert from string to ObjectId:
        document = client.db.collection.find_one({'_id':ObjectId(post_id)})
    
    print collection.find_one({"_id": ObjectId("xkkwow1919")})

##### 批量插入： #####

muilt_data = [{"author": "Mike",
                   "text": "Another post!",
                   "tags": ["bulk", "insert"],
                   "date": datetime.datetime(2009, 11, 12, 11, 14)},
                  {"author": "Eliot",
                   "title": "MongoDB is fun",
                   "text": "and pretty easy too!",
                   "date": datetime.datetime(2009, 11, 10, 10, 45)}]
    
    
    result = collection.insert_many(muilt_data)
    
    print result.inserted_ids

##### 查询多个文档 #####

collection_data = collection.find()
    
    for _data in collection_data:
        print _data

##### 统计 #####

count = collection.count_documents({})
    print count
    T_count = collection.count_documents({"author": "tom"})
    print T_count

##### 组合查询 #####

d = datetime.datetime(2009, 11, 12, 12)
    for post in collection.find({"date": {"$lt": d}}).sort("author"):
        print post

排序：

collection.find().sort("name", pymongo.ASCENDING)
    collection.find().sort([("name", pymongo.ASCENDING), ("active_time", pymongo.DESCENDING)])

##### 创建索引 #####

ind_res = result = db.profiles.create_index([('user_id', pymongo.ASCENDING)], unique=True)
    print list(db.profiles.index_information())

唯一性索引

user_profiles = [{'user_id': 211, 'name': 'Luke'},  {'user_id': 212, 'name': 'Ziltoid'}]
    
    db.profiles.insert_many(user_profiles)
    
    duplicate_profile = {'user_id': 212, 'name': 'Tommy'}
    
    try:
        db.profiles.insert_one(duplicate_profile)
    except:
        print traceback.format_exc()

完整示例代码：

# coding: utf-8
    
    import traceback
    import datetime
    import pymongo
    from pymongo import MongoClient
    from bson.objectid import ObjectId
    
    client = pymongo.MongoClient("mongodb://127.0.0.1:12701/")
    
    db = client.mytest
    
    collection = db['test-collection']
    
    #新建collection 并插入文档
    
    init_data = {"author": "tom", "text": "My first blog post!",
                 "tags": ["mongodb", "python", "pymongo"],
                 "date": datetime.datetime.utcnow()}
    
    post_id = collection.insert_one(init_data).inserted_id
    
    print post_id
    
    # 查看collection是否创建
    print db.list_collection_names()
    print collection.find_one({"_id": post_id})
    
    print collection.find_one({"_id": ObjectId("xkkwow1919")})
    
    # 查找 collection
    print collection.find({"author": "Mike"})
    print collection.find_one({"author": "Mike"})
    
    
    # 插入多个多个文档
    muilt_data = [{"author": "Mike",
                   "text": "Another post!",
                   "tags": ["bulk", "insert"],
                   "date": datetime.datetime(2009, 11, 12, 11, 14)},
                  {"author": "Eliot",
                   "title": "MongoDB is fun",
                   "text": "and pretty easy too!",
                   "date": datetime.datetime(2009, 11, 10, 10, 45)}]
    
    
    result = collection.insert_many(muilt_data)
    
    print result.inserted_ids
    
    
    # 查找多个文档
    # 返回cusor
    collection_data = collection.find()
    
    for _data in collection_data:
        print _data
    
    
    # 统计
    
    count = collection.count_documents({})
    print count
    T_count = collection.count_documents({"author": "tom"})
    print T_count
    
    
    # 组合查询
    
    d = datetime.datetime(2009, 11, 12, 12)
    for post in collection.find({"date": {"$lt": d}}).sort("author"):
        print post
    
    
    # 创建索引
    
    ind_res = result = db.profiles.create_index([('user_id', pymongo.ASCENDING)], unique=True)
    print list(db.profiles.index_information())
    
    
    user_profiles = [{'user_id': 211, 'name': 'Luke'},  {'user_id': 212, 'name': 'Ziltoid'}]
    
    db.profiles.insert_many(user_profiles)
    
    duplicate_profile = {'user_id': 212, 'name': 'Tommy'}
    
    try:
        db.profiles.insert_one(duplicate_profile)
    except:
        print traceback.format_exc()

输出：

[u'test-collection', u'profiles', u'people']
    <pymongo.cursor.Cursor object at 0x03C75810>
    {u'date': datetime.datetime(2019, 7, 31, 8, 20, 42, 595000), u'text': u'My first blog post!', u'_id': ObjectId('5d414f5ac4decf11266bc71c'), u'author': u'Mike', u'tags': [u'mongodb', u'python', u'pymongo']}
    [ObjectId('5d415f266d4b03250b852ba0'), ObjectId('5d415f266d4b03250b852ba1')]
    ....
    ObjectId('5d415ee96e9053b3e893f0d4'), u'author': u'Eliot', u'title': u'MongoDB is fun'}
    {u'date': datetime.datetime(2009, 11, 12, 11, 14), u'text': u'Another post!', u'_id': ObjectId('5d415ef48ece310693da7c23'), u'author': u'Mike', u'tags': [u'bulk', u'insert']}
    
    .....
    {u'date': datetime.datetime(2009, 11, 10, 10, 45), u'text': u'and pretty easy too!', u'_id': ObjectId('5d415ef48ece310693da7c24'), u'author': u'Eliot', u'title': u'MongoDB is fun'}
    ....
    {u'date': datetime.datetime(2009, 11, 12, 11, 14), u'text': u'Another post!', u'_id': ObjectId('5d415f266d4b03250b852ba0'), u'author': u'Mike', u'tags': [u'bulk', u'insert']}
    {u'date': datetime.datetime(2009, 11, 10, 10, 45), u'text': u'and pretty easy too!', u'_id': ObjectId('5d415f266d4b03250b852ba1'), u'author': u'Eliot', u'title': u'MongoDB is fun'}
    12
    0
    {u'date': datetime.datetime(2009, 11, 10, 10, 45), u'text': u'and pretty easy too!', u'_id': ObjectId('5d415ec586f95fede51f2605'), u'author': u'Eliot', u'title': u'MongoDB is fun'}
    {u'date': datetime.datetime(2009, 11, 10, 10, 45), u'text': u'and pretty easy too!', u'_id': ObjectId('5d415f266d4b03250b852ba0'), u'author': u'Mike', u'tags': [u'bulk', u'insert']}
    [u'user_id_1', u'_id_']
    Traceback (most recent call last):
      File "", line 87, in <module>
        db.profiles.insert_one(duplicate_profile)
      File "C:\Python27\lib\site-packages\pymongo\collection.py", line 700, in insert_one
        session=session),
      File "C:\Python27\lib\site-packages\pymongo\collection.py", line 614, in _insert
        bypass_doc_val, session)
      File "C:\Python27\lib\site-packages\pymongo\collection.py", line 602, in _insert_one
        acknowledged, _insert_command, session)
      File "C:\Python27\lib\site-packages\pymongo\mongo_client.py", line 1280, in _retryable_write
        return self._retry_with_session(retryable, func, s, None)
      File "C:\Python27\lib\site-packages\pymongo\mongo_client.py", line 1233, in _retry_with_session
        return func(session, sock_info, retryable)
      File "C:\Python27\lib\site-packages\pymongo\collection.py", line 599, in _insert_command
        _check_write_command_response(result)
      File "C:\Python27\lib\site-packages\pymongo\helpers.py", line 217, in _check_write_command_response
        _raise_last_write_error(write_errors)
      File "C:\Python27\lib\site-packages\pymongo\helpers.py", line 198, in _raise_last_write_error
        raise DuplicateKeyError(error.get("errmsg"), 11000, error)
    DuplicateKeyError: E11000 duplicate key error collection: mytest.profiles index: user_id_1 dup key: { : 212 }

封装示例：

# coding: utf-8
    
    import pymongo
    from bson.objectid import ObjectId
    
    
    class MongoDB(object):
        def __init__(self, ip, port, database):
            self.client = pymongo.MongoClient("mongodb://%s:%s/" %(ip, port))
            self.database = self.client[database]
    
        def get_table(self, table_name):
            return self.database[table_name]
        
        def get(self, tablename, filters=None):
            result = []
            filters = filters or {}
            collection_data = self.get_table(tablename).find(filters)
            for _data in collection_data:
                result.append(_data)
            
            return result
        
        def get_one(self, tablename, filters=None):
            filters = filters or {}
            return self.get_table(tablename).find_one(filters)
        
        def get_by_id(self, tablename, tid):
            return self.get_table(tablename).find_one({"_id": ObjectId(tid)})
        
        
        def create(self, tablename, data):
            post_id = self.get_table(tablename).insert_one(data)
            tid = post_id.inserted_id
            return tid
        
        def create_many(self, tablename, datas):
            result = self.get_table(tablename).insert_many(datas)
            return result.inserted_ids
        
        
        def count(self, tablename, fileters=None):
            fileters = fileters or {}
            T_count = self.get_table(tablename).count_documents(fileters)
            return T_count
        
        
        def create_index(self, tablename, index_col, unique=False):
            '''
            :param tablename: 
            :param index_col: [('user_id', pymongo.ASCENDING)]
            :param unique: 
            :return: 
            '''
            result = self.get_table(tablename).create_index(index_col, unique=unique)
            return result
        
        def get_index_info(self, tablename):
            return list(self.get_table(tablename).index_information())