python3操作MongoDB4.4
2023-04-18 16:54:15 时间
目录
一、MongoDB连接
MongoDBCONN.py
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
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
相关文章
- 四年完成400万行Python代码检查,甚至顺手写了个编译器
- 4种出色的Python测试框架!
- IEEE 2019编程语言排行榜出炉:趋势、开源、职位需求,Python都是第一
- 面向中级 Python 开发人员的 13 个项目构想
- 是时候对 Python2 说“拜拜了”
- 一个 Python 自动化脚本引发的惨案!把公司搞死了?
- 未来明星语言Julia或成Python劲敌
- Visual Studio 2019优化新鲜出炉:C++后端更新汇总
- 1分钟插入10亿行数据!抛弃Python,写脚本请使用Rust
- 节省显存新思路,在 PyTorch 里使用 2 bit 激活压缩训练神经网络
- TIOBE编程语言排行:C语言第一,Python反超Java,挤进第二
- 1.6 万 Star!微软谷歌都在用 Python 性能测试工具
- 7个实用小技巧,提升PyTorch技能,还带示例演示
- 2021年哪些编程语言的薪酬比较高?
- Java开始没落了吗?学Java真的没前途了吗?
- 入职第一天,就和公司后端闹掰了!
- 纯PyTorch语音工具包开源,Kaldi:我压力有点大
- 人脸识别对动画无效,迪士尼打造动画专用人脸识别库
- Python爬虫需要学多久才能掌握?
- Python30岁,先驱:没想到Python这么流行