目录
一、MongoDB连接
二、MongoDB操作
一、MongoDB连接
MongoDBCONN.py
代码语言：javascript复制import pymongo
"""客户端连接"""
def db_client(ip="127.0.0.1",port="27017"):
url = "mongodb://" ip ":" port
client = pymongo.MongoClient(url)
print("client:",client)
return client
"""创建或连接数据库"""
def db_database(client,dbname):
db = client[dbname]
# dblist = client.database_names() ## 获取数据库列表
dblist = client.list_database_names() ## 获取数据库列表
print("dblist:",dblist)
if dbname not in dblist: # 判断数据库是否存在
print("The database has been created!")
return db
"""创建或连接集合"""
def db_collect(db,colname):
collect = db[colname]
# collist = db.collection_names() ## 获取数据库中的所有集合
collist = db.list_collection_names() ## 获取数据库中的所有集合
print("collist:",collist)
if colname not in collist: # 判断集合是否存在
print("The collect has been created!")
return collect
"""测试代码"""
if __name__ == "__main__":
print(".....start.....")
client = db_client() ## 连接对象
db = db_database(client,"dbtest") ## 数据库
collect = db_collect(db,"coltest") ## 集合
mydict = {"name": "RUNOOB", "alexa": "10000", "url": "https://www.runoob.com"}
x = collect.insert_one(mydict) ## 插入单条数据
print(x.inserted_id)
print("......end......")二、MongoDB操作
MongoDBCRUD.py
代码语言：javascript复制import pymongo
import MongoDBCONN
"""INSERT:插入一条数据"""
def insert_one(collect,dict):
result = collect.insert_one(dict) ## 插入单条数据
id = result.inserted_id
print("INSERT id:",id)
return id
"""INSERT:插入多条数据"""
def insert_many(collect,list):
result = collect.insert_many(list)
ids = result.inserted_ids
print("INSERT ids:",ids)
return ids
"""DELETE:删除一条数据"""
def delete_one(collect,dict):
status = collect.delete_one(dict)
print("DELETE status",status)
return status
"""DELETE:删除多条数据"""
def delete_many(collect,dict):
status = collect.delete_many(dict)
print("DELETE status",status)
return status
"""DELETE:删除集合"""
def drop_collect(collect):
status = collect.drop()
print("DELETE status",status)
return status
"""QUERY:查询一条数据"""
def find_one(collect):
data = collect.find_one()
# print(data)
return data
"""QUERY:查询集合中所有数据"""
def find(collect):
list_dict = collect.find()
# for data in list_dict:
# print(data)
return list_dict
"""QUERY:根据指定条件或者字段查询"""
def find(collect,dict=None):
list_dict = collect.find(dict)
# for data in list_dict:
# print(data)
return list_dict
"""QUERY:返回指定条数记录"""
def find(collect,dict=None,limit=None):
list_dict = collect.find(dict).limit(limit)
# for data in list_dict:
# print(data)
return list_dict
"""SORT:对字段进行排序"""
def sort(collect,field,sort=1):
## sort=1：默认升序 sort=-1：降序排序
list_dict = collect.find(field,sort)
# for dict in list_dict:
# print(dict)
return list_dict
"""UPDATE:修改符合的第一条记录"""
def update_one(collect,query_dict,update_dict):
update = collect.update_one(query_dict,update_dict)
print("number of update:",update.modified_count)
return update
"""UPDATE:修改符合的所有记录"""
def update_many(collect,query_dict,update_dict):
update = collect.update_many(query_dict,update_dict)
print("number of update:",update.modified_count)
return update
if __name__ == '__main__':
client = MongoDBCONN.db_client()
db = MongoDBCONN.db_database(client,"dbtest")
collect = MongoDBCONN.db_collect(db,"coltest") ## 集合
pass
