mongodb python api简单入门


from pymongo import MongoClient
from bson.objectid import ObjectId
import os


class MongodbHelper:
    """MongoDB助手

    方法:
        isServerRunning：mongodb服务器服务是否运行
        startServer：启动服务
        login：登录
        logout：登出，退出登录
        getDatabases：当前用户可见的库
        getTables：库中全部的表-集合
        getRowCount：表中全部行数量
        getAllRows：表中全部行
        findOneById：根据id查某行
        findElementsByColumn：根据k-v查符合条件的行
        insertOne：插入一行
        insertMulit：插入多行
        updateOnePswdByUser：有条件的更新一条记录
        updateMulitPswdByUser：使用正则或模糊查询进行更新
        deleteOneByPswd：有条件的删除记录
        deleteMulitByPswd：有条件的删除多条记录
        clearTable：清空表
        dropTable：删除表
        dropDb：删除数据库
    """

    instance = None
    instanced = False

    # 独一份内存空间，确保单例
    # 写法比较固定
    def __new__(cls):
        if cls.instance is None:
            cls.instance = super().__new__(cls)
        return cls.instance

    # 单例中仅一次初始化即可
    def __init__(self):
        if MongodbHelper.instanced:
            return

        MongodbHelper.instanced = True
        print("初始化地址：%s" % id(self))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

from pymongo import MongoClient

from bson.objectid import ObjectId

import os

class MongodbHelper:

"""MongoDB助手

方法:

isServerRunning：mongodb服务器服务是否运行

startServer：启动服务

login：登录

logout：登出，退出登录

getDatabases：当前用户可见的库

getTables：库中全部的表-集合

getRowCount：表中全部行数量

getAllRows：表中全部行

findOneById：根据id查某行

findElementsByColumn：根据k-v查符合条件的行

insertOne：插入一行

insertMulit：插入多行

updateOnePswdByUser：有条件的更新一条记录

updateMulitPswdByUser：使用正则或模糊查询进行更新

deleteOneByPswd：有条件的删除记录

deleteMulitByPswd：有条件的删除多条记录

clearTable：清空表

dropTable：删除表

dropDb：删除数据库

"""

instance = None

instanced = False

# 独一份内存空间，确保单例

# 写法比较固定

def __new__(cls):

if cls.instance is None:

cls.instance = super().__new__(cls)

return cls.instance

# 单例中仅一次初始化即可

def __init__(self):

if MongodbHelper.instanced:

return

MongodbHelper.instanced = True

print("初始化地址：%s" % id(self))


    def isServerRunning(self):
        logs = os.popen('ps -ef|grep mongo').readlines() # 调用终端执行一行shell指令
        for line in logs:
            if 'mongod --dbpath=/Users/moonmen/appspace/mongodb-macos-x86_64-4.4.1/db --logpath=/Users/moonmen/appspace/mongodb-macos-x86_64-4.4.1/log/mongodb.log --port=27017 --logappend --fork --auth' in line:
                return True
        return False

1

2

3

4

5

6

7

8

def isServerRunning(self):

logs = os.popen('ps -ef|grep mongo').readlines() # 调用终端执行一行shell指令

for line in logs:

if 'mongod --dbpath=/Users/moonmen/appspace/mongodb-macos-x86_64-4.4.1/db --logpath=/Users/moonmen/appspace/mongodb-macos-x86_64-4.4.1/log/mongodb.log --port=27017 --logappend --fork --auth' in line:

return True

return False


    def startServer(self):
        cmd = 'mgdb_start.sh'
        logs = os.popen(cmd).readlines()
        for line in logs:
            if "process started successfully" in line:
                return True
        return False

1

2

3

4

5

6

7

8

9

def startServer(self):

cmd = 'mgdb_start.sh'

logs = os.popen(cmd).readlines()

for line in logs:

if "process started successfully" in line:

return True

return False


    def login(self, userName, pswd):
        # 登录方式1：
        # self.client = MongoClient('mongodb://{}:{}@{}:{}/?authSource={}'.format(userName, pswd, "localhost", "27017", "admin"))

        # 登录方式2：
        self.client = MongoClient(host="localhost", port=27017)
        # db = self.client.admin
        self.db = self.client['admin']
        self.db.authenticate(userName, pswd)

    def logout(self):
        self.client.close()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

def login(self, userName, pswd):

# 登录方式1：

# self.client = MongoClient('mongodb://{}:{}@{}:{}/?authSource={}'.format(userName, pswd, "localhost", "27017", "admin"))

# 登录方式2：

self.client = MongoClient(host="localhost", port=27017)

# db = self.client.admin

self.db = self.client['admin']

self.db.authenticate(userName, pswd)

def logout(self):

self.client.close()


    def getDatabases(self):
        return self.client.list_database_names()  # > show dbs

    def getTables(self):
        return self.__getSingleTables('db_test')

    def __getSingleTables(self, dbName):
        # db = self.client.db_test # 用点方法访问库
        db = self.client[dbName]   # 用列表引用方法访问库
        return db.list_collection_names() # > show tables

1

2

3

4

5

6

7

8

9

10

11

12

def getDatabases(self):

return self.client.list_database_names() # > show dbs

def getTables(self):

return self.__getSingleTables('db_test')

def __getSingleTables(self, dbName):

# db = self.client.db_test # 用点方法访问库

db = self.client[dbName] # 用列表引用方法访问库

return db.list_collection_names() # > show tables


    def getRowCount(self):
        # results = tb.find().sort([('key1',pymongo.ASCENDING),('key2', pymongo.ASCENDING)]) #多条件排序
        # results = tb.find().sort('key', pymongo.ASCENDING).skip(2) #分页提取数据
        return self.__getRowCount('db_test', 'tb_test')

    def __getRowCount(self, dbName, tbName):
        return self.client[dbName][tbName].find().count()

    def getAllRows(self):
        return self.__getAll('db_test', 'tb_test')

    def __getAll(self, dbName, tbName):
        cursor = self.client[dbName][tbName].find()
        eles = []
        for ele in cursor:
            eles.append(ele)
        self.client.close_cursor(cursor.cursor_id) # 关闭表的游标
        return eles

    def findOneById(self, _id):
        db = self.client.db_test
        tb = db.tb_test
        return tb.find_one({'_id': ObjectId(_id)})

    def findElementsByColumn(self, col):
        db = self.client.db_test
        tb = db.tb_test
        return tb.find_one(col)

    def blurryFind(self):
        """
            模糊查询

            符号含义示例
                $lt 小于{'age': {'$lt': 20}} age值小于20
                $gt 大于{'age': {'$gt': 20}}
                $lte 小于等于{'age': {'$lte': 20}}
                $gte 大于等于{'age': {'$gte': 20}}
                $ne 不等于{'age': {'$ne': 20}}
                $in 在范围内{'age': {'$in': [20, 23]}}
                $nin 不在范围内{'age': {'$nin': [20, 23]}}
        """
        pass

    def regexFind(self):
        '''
        正则查询

        符号含义示例
            $regex 匹配正则 {'name': {'$regex': '^M.*'}} name以M开头
            $exists 属性是否存在 {'name': {'$exists': True}} name属性存在
            $type 类型判断 {'age': {'$type': 'int'}} age的类型为int
            $mod 数字模操作 {'age': {'$mod': [5, 0]}} 年龄模5余0 
            $where 高级条件查询 {'$where': 'obj.fans_count == obj.follows_count'} 自身粉丝数等于关注数
        '''
        pass

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

def getRowCount(self):

# results = tb.find().sort([('key1',pymongo.ASCENDING),('key2', pymongo.ASCENDING)]) #多条件排序

# results = tb.find().sort('key', pymongo.ASCENDING).skip(2) #分页提取数据

return self.__getRowCount('db_test', 'tb_test')

def __getRowCount(self, dbName, tbName):

return self.client[dbName][tbName].find().count()

def getAllRows(self):

return self.__getAll('db_test', 'tb_test')

def __getAll(self, dbName, tbName):

cursor = self.client[dbName][tbName].find()

eles = []

for ele in cursor:

eles.append(ele)

self.client.close_cursor(cursor.cursor_id) # 关闭表的游标

return eles

def findOneById(self, _id):

db = self.client.db_test

tb = db.tb_test

return tb.find_one({'_id': ObjectId(_id)})

def findElementsByColumn(self, col):

db = self.client.db_test

tb = db.tb_test

return tb.find_one(col)

def blurryFind(self):

"""

模糊查询

符号含义示例

$lt 小于{'age': {'$lt': 20}} age值小于20

$gt 大于{'age': {'$gt': 20}}

$lte 小于等于{'age': {'$lte': 20}}

$gte 大于等于{'age': {'$gte': 20}}

$ne 不等于{'age': {'$ne': 20}}

$in 在范围内{'age': {'$in': [20, 23]}}

$nin 不在范围内{'age': {'$nin': [20, 23]}}

"""

pass

def regexFind(self):

'''

正则查询

符号含义示例

$regex 匹配正则 {'name': {'$regex': '^M.*'}} name以M开头

$exists 属性是否存在 {'name': {'$exists': True}} name属性存在

$type 类型判断 {'age': {'$type': 'int'}} age的类型为int

$mod 数字模操作 {'age': {'$mod': [5, 0]}} 年龄模5余0

$where 高级条件查询 {'$where': 'obj.fans_count == obj.follows_count'} 自身粉丝数等于关注数

'''

pass


    def insertOne(self):
        db = self.client.db_test
        tb = db.tb_test
        tb = db['tb_test']
        return tb.insert_one({"user": "小王", "pswd": "xiaowang123"})

    def insertMulit(self):
        datas = [{"user": "小王{}".format(i), "pswd": "xiaowang12{}".format(i)} for i in range(5)]
        tb = self.client['db_test']['tb_test']
        return tb.insert_many(datas)

1

2

3

4

5

6

7

8

9

10

11

12

def insertOne(self):

db = self.client.db_test

tb = db.tb_test

tb = db['tb_test']

return tb.insert_one({"user": "小王", "pswd": "xiaowang123"})

def insertMulit(self):

datas = [{"user": "小王{}".format(i), "pswd": "xiaowang12{}".format(i)} for i in range(5)]

tb = self.client['db_test']['tb_test']

return tb.insert_many(datas)


    def updateOnePswdByUser(self, u, new_pswd):
        condition = {"user": u}
        db = self.client.db_test
        tb = db.tb_test
        user = tb.find_one(condition)
        user['pswd'] = new_pswd
        # return tb.update_one(condition, user) # 更新符合条件的第一条记录。update参数只接受$开头的符号语句$set、$inc此类
        # return tb.find_one_and_update(condition, user) # find_one_and_update的update参数只接受$开头的符号语句$set、$inc此类
        return tb.find_one_and_replace(condition, user)

    def updateMulitPswdByUser(self, user):
        condition = {'user': {'$regex': '.*{}.*'.format(user)}} # 模糊匹配 如果是空的{}则更新全部条目
        db = self.client.db_test
        tb = db.tb_test
        return tb.update_many(condition, {'$inc': {'age': 1}})  # age必须是int类型

    def dynamicUpdate(self):
        '''
            动态更新

            update修改器（update_one示例）: 
                $inc：变量值1后保存
                    tb.update_one({"age":90}, {"$inc":{"age":1}}}) 将原先的age值90加1
                $set：没有指定项就自动添加，有则更新
                    tb.update_one({"user":"小明"}, {"$set":{"single":true}}}) 给”小明“这一行增加(或更新)一个字段single，值为true
                $unset：删除指定项
                    tb.update_one({"user":"小明"}, {"$unset":{"single":1}}}) 查找”小明“这一行，删除字段single
                $push：在列表属性(list)的项-字段尾端加入新元素
                    tb.update_one({"user":"小明"}, {"$push":{"like_list":"动画"}}}) 在like_list这个列表中插入一个‘动画’
                $pull：删除列表属性的项目中的指定元素-不限位置和数量
                    tb.update_one({"user":"小明"}, {"$pull":{"like_list":"动画"}}}) 将like_list这个列表中全部的‘动画’删除
                $pop：删除列表属性项目中的第一个（1）或最后一个（-1）元素
                    tb.update_one({"user":"小明"}, {"$pop":{"like_list":1}}}) 在like_list这个列表中第一个‘动画’移除
                $addToSet：向列表字段插入元素
                    tb.update({"user":"小明"},{"$addToSet":{"like_list":['电影', '绘画', '游泳']}}) 向like_list插入多个爱好
                $push、$addToSet参数$each、$slice、$sort（$slice"、"$sort"须配合"$each"使用）：
                    插入随机一个元素
                        tb.update_one({"user":"小明"},{"$addToSet":{"like_list":{"$each":['电影', '绘画', '游泳']}}}) 向like_list插入一个爱好
                    插入后保留末尾2个元素。$slice不能用在addToSet
                        tb.update_one({"user":"小明"},{"$push":{"like_list":{"$each":['电影', '绘画', '游泳'], "$slice":-2}}})
                    排序，插入，保留最后2个元素
                        tb.update_one({"user":"小明"},{"$push":{"like_list":{"$each":['电影', '绘画', '游泳'], "$slice":-2, "$sort":-1}}})
        '''

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

def updateOnePswdByUser(self, u, new_pswd):

condition = {"user": u}

db = self.client.db_test

tb = db.tb_test

user = tb.find_one(condition)

user['pswd'] = new_pswd

# return tb.update_one(condition, user) # 更新符合条件的第一条记录。update参数只接受$开头的符号语句$set、$inc此类

# return tb.find_one_and_update(condition, user) # find_one_and_update的update参数只接受$开头的符号语句$set、$inc此类

return tb.find_one_and_replace(condition, user)

def updateMulitPswdByUser(self, user):

condition = {'user': {'$regex': '.*{}.*'.format(user)}} # 模糊匹配如果是空的{}则更新全部条目

db = self.client.db_test

tb = db.tb_test

return tb.update_many(condition, {'$inc': {'age': 1}}) # age必须是int类型

def dynamicUpdate(self):

'''

动态更新

update修改器（update_one示例）:

$inc：变量值1后保存

tb.update_one({"age":90}, {"$inc":{"age":1}}}) 将原先的age值90加1

$set：没有指定项就自动添加，有则更新

tb.update_one({"user":"小明"}, {"$set":{"single":true}}}) 给”小明“这一行增加(或更新)一个字段single，值为true

$unset：删除指定项

tb.update_one({"user":"小明"}, {"$unset":{"single":1}}}) 查找”小明“这一行，删除字段single

$push：在列表属性(list)的项-字段尾端加入新元素

tb.update_one({"user":"小明"}, {"$push":{"like_list":"动画"}}}) 在like_list这个列表中插入一个‘动画’

$pull：删除列表属性的项目中的指定元素-不限位置和数量

tb.update_one({"user":"小明"}, {"$pull":{"like_list":"动画"}}}) 将like_list这个列表中全部的‘动画’删除

$pop：删除列表属性项目中的第一个（1）或最后一个（-1）元素

tb.update_one({"user":"小明"}, {"$pop":{"like_list":1}}}) 在like_list这个列表中第一个‘动画’移除

$addToSet：向列表字段插入元素

tb.update({"user":"小明"},{"$addToSet":{"like_list":['电影', '绘画', '游泳']}}) 向like_list插入多个爱好

$push、$addToSet参数$each、$slice、$sort（$slice"、"$sort"须配合"$each"使用）：

插入随机一个元素

tb.update_one({"user":"小明"},{"$addToSet":{"like_list":{"$each":['电影', '绘画', '游泳']}}}) 向like_list插入一个爱好

插入后保留末尾2个元素。$slice不能用在addToSet

tb.update_one({"user":"小明"},{"$push":{"like_list":{"$each":['电影', '绘画', '游泳'], "$slice":-2}}})

排序，插入，保留最后2个元素

tb.update_one({"user":"小明"},{"$push":{"like_list":{"$each":['电影', '绘画', '游泳'], "$slice":-2, "$sort":-1}}})

'''


    def deleteOneByPswd(self, pswd):
        db = self.client.db_test
        tb = db.tb_test
        # tb.delete_one({"pswd": pswd})
        tb.find_one_and_delete({"pswd": pswd})

    def deleteMulitByPswd(self, pswd):
        db = self.client.db_test
        tb = db.tb_test
        tb.delete_many({'pswd': {'$regex': '.*{}.*'.format(pswd)}}) # 如果参数传入'123'，删除密码里带有‘123’字样的
    
    def clearTable(self):
        db = self.client.db_test
        tb = db.tb_test
        tb.delete_many({}) # 条件为空

    def clearTable2(self):
        tb = self.client['db_test']['tb_test']
        tb.remove() # tb.delete_many({})

    def dropTable(self):
        tb = self.client['db_test']['tb_test']
        tb.drop() # 删除表
    
    def dropDb(self):
        self.client.drop_database('db_test') # 删除数据库

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

def deleteOneByPswd(self, pswd):

db = self.client.db_test

tb = db.tb_test

# tb.delete_one({"pswd": pswd})

tb.find_one_and_delete({"pswd": pswd})

def deleteMulitByPswd(self, pswd):

db = self.client.db_test

tb = db.tb_test

tb.delete_many({'pswd': {'$regex': '.*{}.*'.format(pswd)}}) # 如果参数传入'123'，删除密码里带有‘123’字样的

def clearTable(self):

db = self.client.db_test

tb = db.tb_test

tb.delete_many({}) # 条件为空

def clearTable2(self):

tb = self.client['db_test']['tb_test']

tb.remove() # tb.delete_many({})

def dropTable(self):

tb = self.client['db_test']['tb_test']

tb.drop() # 删除表

def dropDb(self):

self.client.drop_database('db_test') # 删除数据库


from mongodb_helper import MongodbHelper

helper = MongodbHelper()

if not helper.isServerRunning():
    helper.startServer()
helper.login("root", "root")

print(helper.getDatabases())
print(helper.getTables())
print(helper.getRowCount())
print(helper.getAllRows())
helper.insertOne()
helper.insertMulit()
helper.updateOnePswdByUser('小a', 'haha')
print(helper.findOneById("5f9187ad5ea3f3329a16d09f"))
print(helper.findElementsByColumn({'user': '小王'}))
print(helper.updateOnePswdByUser('小明', '999999'))
print(helper.updateMulitPswdByUser('小明'))
helper.deleteMulitByPswd('123')
helper.dropTable()
helper.dropDb()
helper.logout()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

from mongodb_helper import MongodbHelper

helper = MongodbHelper()

if not helper.isServerRunning():

helper.startServer()

helper.login("root", "root")

print(helper.getDatabases())

print(helper.getTables())

print(helper.getRowCount())

print(helper.getAllRows())

helper.insertOne()

helper.insertMulit()

helper.updateOnePswdByUser('小a', 'haha')

print(helper.findOneById("5f9187ad5ea3f3329a16d09f"))

print(helper.findElementsByColumn({'user': '小王'}))

print(helper.updateOnePswdByUser('小明', '999999'))

print(helper.updateMulitPswdByUser('小明'))

helper.deleteMulitByPswd('123')

helper.dropTable()

helper.dropDb()

helper.logout()

威格灵

1.判断mongo服务是否启动

2.启动服务

3.登录登出

4.库和表信息

5.查

6.增

7.改

8.删

9.测试例句