Flask Restful 接口 - SqlAlchemy & Marshmallow

公冶龙野

2023-12-01

本篇通过一个一对多的具体示例，介绍如何使用Flask，Flask-SQLAlchemy，marshmallow-sqlalchemy构建restful风格的api

准备阶段

├── app.py

├── data.db

├── models

│ ├── models.py ---> 用于数据库表的映射

├── resources

├── schema

│ └── schemas.py ---> 用于Json序列化与反序列化

安装依赖

pip install Flask

pip install Flask-SQLAlchemy

pip install marshmallow

pip install marshmallow-sqlalchemy

app.py初始化

#导入必要的包
from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.db' #加载数据库配置
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app) #创建db
api = Api(app)


if __name__ == "__main__":
    db = SQLAlchemy(app)
    app.run(debug=True)

建立一对多关系模型 - Flask SqlAlchemy

通过一个商店(Store)可以有多个商品(Item)来演示

Step1:创建数据库table

CREATE TABLE t_store (  
    id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
    name TEXT
);

CREATE TABLE t_item (  
    id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
    name TEXT,
  	price FLOAT,
  	storeid INTEGER,
    FOREIGN KEY(storeid) REFERENCES t_store(id),
);

Step2 : 声明Item模型并创建外建

外建关联两者，在多的一边声明

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

class ItemModel(db.Model):

    __tablename__ = 't_item'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), unique=True)
    price = db.Column(db.Integer)
    storeid = db.Column(db.Integer,db.ForeignKey('t_store.id')) #外建关联
    
    def __init__(self,name,price,storeid):
        self.name = name
        self.price = price
        self.storeid = storeid

    def save_to_db(self):  #添加到数据库中 -- 后面resource类会使用到
        db.session.add(self)
        db.session.commit()

    def delete_from_db(self): #从数据库中删除该记录
        db.session.delete(self)
        db.session.commit()

Step3 : 声明Store模型并定义关系属性
- 关系属性在一的那一方定义
- 关系属性是用来表示一的这一方拥有多那一方的一个属性

class StoreModel(db.Model):

    __tablename__ = 't_store'

    id = db.Column(db.Integer, primary_key=True)
    store_name = db.Column('store_name',db.String(80), unique=True)
    items = db.relationship('ItemModel',backref='store')
    
    def __init__(self,store_name):
        self.store_name = store_name
    
    def save_to_db(self):
        db.session.add(self)
        db.session.commit()

    def delete_from_db(self):
        db.session.delete(self)
        db.session.commit()

    def __repr__(self):
        return self.store_name

backref='store'会自动在ItemModel中建立了store的反向关系属性表示商品属于哪一个商店

序列化与反序列化 - Marshmallow

模型声明完成后，下一步可以去创建Schema类来处理Json的序列化与反序列化

先建立Item的Schema

from models import models
from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
from marshmallow import fields,post_load

class ItemSchema(SQLAlchemySchema):

    id = auto_field(dump_only=True) #read_only字段
    name = auto_field()
    price = fields.Float()
    storeid = auto_field(load_only=True) #write_only字段
    store = fields.Str()
    
    #反序列化
    @post_load
    def make_item(self, data, **kwargs):
        return models.ItemModel(**data)

    class Meta:
        model = models.ItemModel

model = models.ItemModel：用于和模型类关联，关联后会根据ItemModel的属性创建对应的json的key
dump_only=True：表示id是只读字段，只允许从模型序列化。
load_only=True：表示storeid是可只能写入，只允许根据请求体反序列化，一般在post或put接口中使用
@post_load方法: 反序列化，POST请求时通过该方法创建itemModel的实例并返回

建立Store的Schema

class StoreSchema(SQLAlchemySchema):
    id = auto_field(dump_only=True)
    store_name = auto_field()
    #返回items的列表
    items = fields.Nested(ItemSchema,many=True,dump_only=True,only=('id','name','price'))

    @post_load
    def create_store(self, data, **kwargs):
        return models.StoreModel(**data)
    
    class Meta:
        model = models.StoreModel

fields.Nested(ItemSchema,many=True,dump_only=True,only=('id','name','price'))：返回一个list，包含了所关联商品(Item)的JsonSchema,并且移除store字段避免套娃

创建接口 - Flask Restful

分别创建商店和商品的增删改查接口，商品在新增时须提供storeid用于关联商店，接口列表

Store：

GET /store/{id}
post /store
put /store/{id}
delete /store/{id}

Item：

GET /item/{id}
post /item
put /item/{id}
delete /item/{id}

在resources目录下新建item.py

from flask_restful import Resource, reqparse
from models.models import ItemModel, StoreModel
from schema import schemas
item_schema = schemas.ItemSchema()

class Item(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('name',type=str,required=True,help="Every item needs a name.")

    def get(self, id):
        item = ItemModel.query.filter_by(id=id).first_or_404()
        return item_schema.dump(item)

    def post(self):
        data = self.parser.parse_args()
        if not StoreModel.query.filter_by(id=data['storeid']).first():
            return {'message':'商店不存在'},400

        item = item_schema.load(data)
        try:
            item.save_to_db()
        except:
            return {"message": "An error occurred inserting the item."}, 500

        return item_schema.dump(item),201

    def delete(self, id):
        item = ItemModel.fquery.filter_by(id=id).first_or_404()
        item.delete_from_db()
        return {'message': 'Item deleted.'},200

    def put(self, id):
        data = Item.parser.parse_args()
        item = ItemModel.query.filter_by(id=id).first_or_404()

        for column in item.__table__.columns:
            if column.name != 'id': setattr(item,column.name,data.get(column.name))

        item.save_to_db()
        return item_schema.dump(item)

在resources目录下新建store.py

from flask_restful import Resource, reqparse
from models.models import StoreModel
from schema import schemas
store_schema = schemas.StoreSchema()

class Store(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('store_name',type=str,required=True,help="Every store needs a name.")

    def get(self, id):
        store = StoreModel.query.filter_by(id=id).first_or_404()
        return store_schema.dump(store),200

    def post(self):
        data = self.parser.parse_args()
        store = store_schema.load(data)

        try:
            store.save_to_db()
        except:
            return {"message": "An error occurred"}, 500

        return store_schema.dump(store), 201

    
    def delete(self, id):
        StoreModel.query.filter_by(id=id).first_or_404()
        return {'message': 'Store deleted.'},200
    
    def put(self, id):
        data = self.parser.parse_args()
        store = StoreModel.query.filter_by(id=id).first_or_404()

        store.store_name = data['store_name']
        store.save_to_db()

        return store_schema.dump(store),200

在app.py中使用 api.add_resource 方法注册资源并定义路由

#先导入Item和Store类
from resources.item import Item
from resources.store import Store

api.add_resource(Item,'/item/<int:id>')
api.add_resource(Item,'/item',endpoint='create_item',methods=['POST'])
api.add_resource(Store,'/store/<int:id>')
api.add_resource(Store,'/store',endpoint='create_store',methods=['POST'])

Flask Restful 接口 - SqlAlchemy & Marshmallow

准备阶段

建立一对多关系模型 - Flask SqlAlchemy

序列化与反序列化 - Marshmallow

创建接口 - Flask Restful

相关阅读

相关文章

相关问答

相关文档