python使用flask框架操作mongodb基本使用

python使用flask框架操作mongodb基本使用

系统 vm虚拟机 + ubuntu18.04 python版本3.6.9 Flask版本2.0.3

上一节 mongodb基本使用（python） 转自 清夢懮

本节 简单讲一下如何简单用框架操作mongodb数据库，大部分都是上一节的内容只是改了改，变成一个框架操作数据库了而已

用什么框架不重要重要的是明白理解学以致用！

进入正题:

        先下载flask要是觉得怕把环境搞坏了就创建一个虚拟环境 

ubuntu创建python虚拟环境  转自 清夢懮

        然后使用pip3 install Flask下载flask要是说pip3没找到就重新下载sudo apt instlal pip3 应该是

        然后再下载flask，下载完之后就能正常的使用了，我先把上次的文件挪了过来整理了目录结构如

 然后就改写里面的详细的东西，创建了一个统一测试的接口自己测试测了半天才写的差不多

我们先编写main入口文件

fig import flask_default
from flask_cors import CORS
#  操作数据库文件
_do import mongo_viewfrom flask import Flask# 实例化flask对象
app = Flask(__name__)fig['JSON_AS_ASCII'] = False
# 跨域
CORS(app,cors_allowed_origins="*")# views  数据库操作蓝图
ister_blueprint(mongo_view)if __name__ == '__main__':app.run(host=flask_default['host'],port=flask_default['port'])

主路口的配置文件

mongo_client={# ip"host":"127.0.0.1",# 端口"port":27017,# 账号"username":"admin",# 密码"pwd":"admin",# 连接的库"database":"sange"
}flask_default = {# 端口"port":"6379",# ip"host":"127.0.0.1"
}

 跟那些框架差不多都是实例化、中间件、蓝图、run，然后run里面写一堆的ip，端口啥的

然后写蓝图

from flask import Blueprint, jsonify, request
from flask.views import  MethodView
from mongo_db import dbmongo_view = Blueprint("mongo_view",__name__)#  主页
class Hello(MethodView):def get(self):print("")return "Hello"# 操作mongocleardb的视图
class Params_server(MethodView):# 获取参数，type为one获取一个空为全部def get(self):type = ("type",None)find_where = ("find_where",None)if type == "one":return  jsonify({"err":"200","data":str(db.params_find_one(find_where=find_where)),"msg":"查找成功"}) else:return  jsonify({"err":"200","data":str(db.params_find_all(find_where=find_where)),"msg":"查找成功"})  #  添加一个或者多个数据def post(self):data = ("data",None)print(data,type(data))if not data:return jsonify({"err":"205","msg":"请输入要添加的参数"})else:try:db.add_one(data = data)except Exception as e:return jsonify({"err":"205","msg":"添加失败,请确认格式等信息"+str(e)})return jsonify({"err":"200","msg":"添加成功"})# 更新参数def put(self):data = ("data",None)find_where = ("find_where",None)print(data,find_where)if not (find_where and data):return jsonify({"err":"205","data":"请输入需更新的参数或更新后的数据"})else:try:db.params_update_one(find_where=find_where,update_data=data)    except Exception as e:return jsonify({"err":"205","msg":"更新出错"+str(e)})return jsonify({"err":"205","msg":"更新成功"})# 主页
mongo_view.add_url_rule('/',view_func= Hello.as_view("Hello"))
# 数据库交互
mongo_view.add_url_rule("/mongodb",view_func=Params_server.as_view("Params_server"))

主页的那个只是展示主页显示的数据，用来测试的，为了测试逻辑更代码是否能通顺的 如 

这样就能简单的实现了操作数据库

 其次是操作数据库的代码 mongo_db.py

db_client import MongoDBLink
fig import mongo_client
db = MongoDBLink(mongo_client['host'],mongo_client['port'],mongo_client['username'],mongo_client['pwd'],database=mongo_client['database'])if __name__ == "__main__":data = {'title' : 'titile1 ' ,'about':'123123',}print(db.params_find_one(database='test',find_where=data))

mongodb_client.py

import json
import pymongo# 一个简单的序列化器 class number_dict(object):def __init__(self,data):self.data = datadef to_list(self):list1 = []for i in self.data:list1.append(i)return list1# 启动类 
class MongoDBLink(object):#  需要四个参数启动地址、端口号、账号、密码、连接的库def __init__ (self,host,port,username,pwd,database):self.host = hostself.port = portself.client = pymongo.MongoClient(host = self.host,port=self.port,username=username,password=pwd,)self.db = self.client[database]self.data = database# 添加一个数据 # 参数，json类型的数据，数据库名字def add_one(self,data):data = json.loads(data)print(data,type(data))self.db[self.data].insert_one(dict(data))# try:#     print(data)#     print(self.data)#     self.db[self.data].insert_one(data)# except Exception as e:#     print("添加数据时候出错",e)#     return False# print("添加数据成功{}".format(data))# return True# 查找一个数据  find_where查询条件默认查全部def params_find_one(self,find_where=None):if find_where:return   self.db[self.data].find_one(find_where)else:return self.db[self.data].find_one()#  查找全部数据 find_where查询条件默认查全部def params_find_all(self,find_where=None):print(find_where)# data = self.db[database].find()if find_where:return number_dict(self.db[self.data].find(find_where)).to_list()else:return number_dict(self.db[self.data].find()).to_list()#  更新数据库指定的参数def params_update_one(self,update_data,find_where):print(update_data,find_where)self.db[self.data].update_one(json.loads(find_where),{"$set":json.loads(update_data)})# def params

通过这几样的配置就能够操作数据库了

然后是api测试的文件，为了方便测试就单独写了一个文件创建一个类，

import requests
import jsonurl  = '127.0.0.1:6379/mongodb?type=one'# 接口测试
import requests# 发起http请求
class HttpApiTest:# 修改接口def test_put(self, url, data={}):print(data)res = requests.put(url, data=data)# 插入接口def test_post(self, url, data={}):print(data)res = requests.post(url, data=data)# 查询接口def test_get(self, url, data={}):res = (url, params=data)# 删除接口def test_del(self, url, data={}):res = requests.delete(url, params=data)if __name__ == '__main__':# 实例化对象httpapi = HttpApiTest()# 请求接口# res = st_post("localhost:5000/add_user/", data={"email":"123","password":""}) post_number = json.dumps({"name":"zhangsan","value":"18","height":1.8,"age":"18","sex":"man"})updata_number =  json.dumps({"name":"lisi","value":"18","height":1.8,"age":"18","sex":"man"})find_where  =json.dumps({"name":"zhangsan"})# 获取res = st_get("127.0.0.1:6379/mongodb", data={"find_where":""})# 添加# res = st_post("127.0.0.1:6379/mongodb", data={"data":post_number})# 更新# res = st_put("127.0.0.1:6379/mongodb", data={"data":updata_number,"find_where":find_where})# print(json.loads(res)['data'])print(res)

源码--网盘 转自 清夢懮