python-mongodb簡單封裝

#!/usr/bin/python
# -*- coding: UTF-8 -*-
'''@auther :mr.qin
@IDE:pycharm'''


import pymongo
from tool.Common import *
import sys

class Operation_Mongo(object):
    def __init__(self,db='creeper_test'):
        '''初始化連接'''
        self.connect_client=pymongo.MongoClient("mongodb://{}:{}@{}:27017/".format('xx','xx',tencent_cloud_host))
        self.mydb = self.connect_client[db]#連接指定數據庫

    def insert_collection(self,collection_name,value):#單個插入
        mycol=self.mydb[collection_name]
        mycol_id=mycol.insert_one(value)
        return mycol_id.inserted_id #返回insert_id，即插入文檔的id值

    def insert_batch_collection(self,collection_name,value_list):#批量插入
        mycol=self.mydb[collection_name]
        mycol_id=mycol.insert_many(value_list)
        return mycol_id.inserted_ids #返回insert_id集合，即插入文檔的id值


    def select_one_collection(self,collection_name,search_col=None):#獲取一條數據
        '''search_col：只能是dict類型,key大於等於一個即可，也可為空
        可使用修飾符查詢：{"name": {"$gt": "H"}}#讀取 name 字段中第一個字母 ASCII 值大於 "H" 的數據
        使用正則表達式查詢：{"$regex": "^R"}#讀取 name 字段中第一個字母為 "R" 的數據'''
        my_col=self.mydb[collection_name]
        try:
            result = my_col.find_one(search_col)  # 這里只會返回一個對象，數據需要自己取
            return result
        except TypeError as e:
            print('查詢條件只能是dict類型')
            return None

    def select_all_collection(self,collection_name,search_col=None,limit_num=sys.maxsize,sort_col='None_sort',sort='asc'):
        '''search_col：只能是dict類型,key大於等於一個即可，也可為空
        可使用修飾符查詢：{"name": {"$gt": "H"}}#讀取 name 字段中第一個字母 ASCII 值大於 "H" 的數據
        使用正則表達式查詢：{"$regex": "^R"}#讀取 name 字段中第一個字母為 "R" 的數據
        limit_num:返回指定條數記錄，該方法只接受一個數字參數(sys.maxsize:返回一個最大的整數值)'''
        my_col=self.mydb[collection_name]
        try:
            if sort_col==False or sort_col=='None_sort':
                results=my_col.find(search_col).limit(limit_num)#這里只會返回一個對象，數據需要自己取
            else:
                sort_flag = 1
                if sort == 'desc':
                    sort_flag = -1
                results = my_col.find(search_col).sort(sort_col,sort_flag).limit(limit_num)  # 這里只會返回一個對象，數據需要自己取
            result_all=[i for i in results]#將獲取到的數據添加至list
            return result_all
        except TypeError as e:
            print('查詢條件只能是dict類型')
            return None

    def update_one_collecton(self,collection_name,search_col,update_col):
        '''該方法第一個參數為查詢的條件，第二個參數為要修改的字段。
            如果查找到的匹配數據多余一條，則只會修改第一條。
            修改后字段的定義格式： { "$set": { "alexa": "12345" } }'''
        my_col=self.mydb[collection_name]
        try:
            relust=my_col.update_one(search_col,update_col)
            return relust
        except TypeError as e:
            print('查詢條件與需要修改的字段只能是dict類型')
            return None

    def update_batch_collecton(self,collection_name,search_col,update_col):
        '''批量更新數據'''
        my_col=self.mydb[collection_name]
        try:
            relust=my_col.update_many(search_col,update_col)
            return relust
        except TypeError as e:
            print('查詢條件與需要修改的字段只能是dict類型')
            return None

    def delete_one_collection(self,collection_name,search_col):#刪除集合中的文檔
        my_col = self.mydb[collection_name]
        try:
            relust=my_col.delete_one(search_col)
            return relust
        except TypeError as e:
            print('查詢條件與需要修改的字段只能是dict類型')
            return None


    def delete_batch_collection(self,collection_name,search_col):#刪除集合中的多個文檔
        '''刪除所有 name 字段中以 F 開頭的文檔:{ "name": {"$regex": "^F"} }
        刪除所有文檔：{}'''
        my_col = self.mydb[collection_name]
        try:
            relust=my_col.delete_many(search_col)
            return relust
        except TypeError as e:
            print('查詢條件與需要修改的字段只能是dict類型')
            return None


    def drop_collection(self,collection_name):
        '''刪除集合，如果刪除成功 drop() 返回 true，如果刪除失敗(集合不存在)則返回 false'''
        my_col = self.mydb[collection_name]
        result=my_col.drop()
        return result

    def get_connections(self):#獲取所有的connections
        return self.mydb.list_collection_names()

    def close_connect(self):
        self.connect_client.close()
        return 'mongo連接已關閉'



if __name__=="__main__":
    om=Operation_Mongo()
    dict={ "name": "RUNOOB", "alexa": "10000", "url": "https://www.runoob.com" }
    my_dict_list=[
  { "name": "Taobao", "alexa": "100", "url": "https://www.taobao.com" },
  { "name": "QQ", "alexa": "101", "url": "https://www.qq.com" },
  { "name": "Facebook", "alexa": "10", "url": "https://www.facebook.com" },
  { "name": "知乎", "alexa": "103", "url": "https://www.zhihu.com" },
  { "name": "Github", "alexa": "109", "url": "https://www.github.com" }
]#不指定id
    my_dict_list_assign_id = [
  { "_id": 1, "name": "RUNOOB", "cn_name": "菜鳥教程"},
  { "_id": 2, "name": "Google", "address": "Google 搜索"},
  { "_id": 3, "name": "Facebook", "address": "臉書"},
  { "_id": 4, "name": "Taobao", "address": "淘寶"},
  { "_id": 5, "name": "Zhihu", "address": "知乎"}
]#指定id
    # batch_in=om.insert_batch_collection('test3',my_dict_list_assign_id)
    # print(batch_in)
    # a=om.create_colltion('test2',dict)
    # myquery = {"name": {"$gt": "H"}}#可使用修飾符查詢
    # myquery_2={ "name": { "$regex": "^\W" } }
    # data=om.select_all_collection('test3',sort_col='name',sort='desc')
    # data=om.select_all_collection('test3')
    # om.delete_one_collection('test3',search_col={ "_id": 1, "name": "RUNOOB", "cn_name": "菜鳥教程"})
    # om.delete_batch_collection('test3',search_col={})
    # data = om.select_all_collection('test3')
    # datas=om.select_all_collection('test3',myquery_2)
    # newvalues = {"$set": {"name": "{}_1"}}
    # up_data=om.update_one_collecton('test3',myquery_2,newvalues)
    # bup_data = om.update_batch_collecton('test3', myquery_2, newvalues)
    # data = om.select_all_collection('test3', myquery_2)
    print(om.get_connections())
    # print(om.drop_collection('test2'))
    om.close_connect()
    print(om.get_connections())