[Python知识库]使用flasksqlalchemy定义一个简单的接口

使用的数据库为：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请求的数据写入：

?再次查看数据库中的表（已经写入到数据库了）：

?