使用的数据库为：mysql，
选择的表的数量：1，（表名：Orders）
已经实现的请求方式和功能：
get请求：读取，筛选数据
post请求：插入的方式更新数据
文件目录情况：
程序入口为app.py,可通过python脚本运行，也可以写一个服务，再systemctl服务化启动
app文件内容：
from flask import Flask
from database import mysqldb
from flask_restful import Api
from resources.routes import initialize_routes
from flask_cors import CORS
from resources import errors
app = Flask(__name__)
CORS(app)
app.config.from_pyfile('databaseinfo.py')
#初始化mysql数据库
mysqldb.db.initialize_db(app)
#初始化flask_restful接口
api = Api(app,errors=errors)
#初始化资源，绑定url到api
initialize_routes(api)
if (__name__ == '__main__'):
app.run(debug=False,host="0.0.0.0",port=4000)
# %%
I.其中使用的databaseinfo.py存放着数据库的连接信息，数据库的密码和IP已使用**隐藏
databaseinfo.py文件内容如下：
from flask import app
SQLALCHEMY_DATAbase_URI = 'mysql+pymysql://root:*******@***.***.**.***:3306/Gadaite'
#是否追踪对象修改，用于Flask-SQLalchemy的事件通知系统
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_POOL_SIZE = 30 #解决数据库连接数的问题
SQLALCHEMY_MAX_OVERFLOW = 60
SQLALCHEMY_BINDS = {
'Gadaite': SQLALCHEMY_DATAbase_URI
}
II.其中初始化化mysql数据库的db.py文件如下：
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
def initialize_db(app):
db.init_app(app)
with app.app_context():
db.create_all()
return db
routes.py中声明初始化接口：
from resources.orders import OrdersApi
def initialize_routes(api):
api.add_resource(OrdersApi,'/orders')
显然接口为：/orders 接口类为：OrdersApi
接口类定义在orders.py文件中，文件中定义了get，post，delete请求方式
orders.py文件如下：
from flask_restful import Resource,reqparse
from database.mysqldb import db
from flask import request
from database.mysqldb.models import OrdersModel
from flask import Flask ,render_template
import json
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
#connection = create_engine("mysql+pymysql://root:*******@***.***.**.***:3306/Gadaite")
Session = sessionmaker(bind=create_engine("mysql+pymysql://root:*******@***.***.**.***:3306/Gadaite"))
session = Session()
class OrdersApi(Resource):
##使用get方式请求数据库并返回相关表字段筛选结果
def __init__(self) -> None:
super().__init__()
def get(self):
#传递
parser = reqparse.RequestParser()
parser.add_argument('order_num')
parser.add_argument('order_date')
parser.add_argument('cust_id')
#解析
args = parser.parse_args()
order_num = args['order_num']
order_date = args['order_date']
cust_id = args['cust_id']
#筛选
filters = OrdersModel.query
if order_num is not None:
filters = filters.filter_by(order_num = order_num)
if order_date is not None:
filters = filters.filter_by(order_date = order_date)
if cust_id is not None:
filters = filters.filter_by(cust_id = cust_id)
#返回
rows = filters.all()
if len(rows)==0:
return [{
"order_num":None,
"order_date":None,
"cust_id":None
}]
else:
return [row.dictRepr() for row in rows]
##使用post方式增加相关信息写入数据库
def post(self):
#传递
parser = reqparse.RequestParser()
parser.add_argument('order_num')
parser.add_argument('order_date')
parser.add_argument('cust_id')
#解析
args = parser.parse_args()
order_num = args['order_num']
order_date = args['order_date']
cust_id = args['cust_id']
#db=SQLalchemy()
#body = request.get_json()
#ordersmodel = OrdersModel(**body)
#db.session.add(ordersmodel)
#db.session.commit()
ordersmodel = OrdersModel(order_num = order_num,order_date = order_date,cust_id = cust_id)
session.add(ordersmodel)
session.commit()
#session.close()
return {'order_num': ordersmodel.order_num,'order_date': ordersmodel.order_date,"cust_id":ordersmodel.cust_id}, 200
def delete(self):
OrdersModel.query.delete()
db.session.commit()
return "succeed to delete", 200
接下来是接口类中使用到的mysql数据库的表的ORM类，这里计划是将所有的表的类放在同一个model.py里面，所以model.py中包含Orders表对应的OrdersModel类
model.pywen文件内容如下：
from database.mysqldb.db import db
class OrdersModel(db.Model):
__tablename__ = 'Orders'
order_num = db.Column(db.String(50),unique = True,primary_key = True)
order_date = db.Column(db.String(50),nullable = False,default = False)
cust_id = db.Column(db.String(50),nullable = False,default = False)
def dictRepr(self):
info ={
"order_num":self.order_num,
"order_date":self.order_date,
"cust_id":self.cust_id
}
return info
计划errors.py中用于存放异常处理信息，此处为空的文件夹。
接下来运行文件python app.py
linux终端信息如下：
C(SSpark) [root@VM-16-5-centos flask_Customers]# python app.py * Serving Flask app 'app' (lazy loading) * Environment: production WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Debug mode: off * Running on all addresses. WARNING: This is a development server. Do not use it in a production deployment. * Running on http://***.***.**.***:4000/ (Press CTRL+C to quit)
原始数据库表的信息如下：
使用post工具分别测试 get请求和post请求
get请求：(不带参数查看全部，使用参数进行筛选)
接下来是post请求的数据写入：
再次查看数据库中的表（已经写入到数据库了）：
