项目上手
2023年8月3日大约 6 分钟
项目结构
flask本身并不提供项目结构，框架本身只是提供一个最基本的APP。
因此，在项目开发时，需要先自己搭建项目：
可以按照功能模块划分为APP，当然，flask中称为蓝图，类似 django，如下所示
flaskProject
├── app
│ ├── __init__.py # 创建APP，挂载所有配置、第三方插件、路由
│ ├── extensions.py # 配置第三方插件
│ └── config.py # 项目基本配置
├── goods # 蓝图，类似django的app
│ ├── __init__.py # 创建蓝图对象
│ ├── models.py # 创建模型类
│ ├── fields.py # 指定模型的序列化字段
│ └── views.py # 视图处理
├── manage.py # 项目的管理文件
└── sqlite.db # 数据库，也可以使用mysql数据库基本组件
第三方插件
在flask项目中，需要使用：
flask_sqlalchemy数据库管理插件flask_migrate模型类迁移插件flask_cors跨域插件mysqlclient或pymysqlmysql数据库连接插件
app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_cors import CORS
# import pymysql
# pymysql.install_as_MySQLdb()
# 使用mysql数据库时，要么使用pymysql，然后这么配置
# 要么直接使用 mysqlclient
db = SQLAlchemy() # 创建db对象，方便后期模型类的创建
migrate = Migrate() # 创建数据库迁移对象
cors = CORS() # 创建跨域对象
def config_extensions(app):
db.init_app(app=app)
migrate.init_app(app=app, db=db)
cors.init_app(app)配置文件
app/settings.py
import base64
import os
# 通过如下方式生成secret_key，然后将生成的密钥添加到配置文件，不要泄露密钥
# secret_key = base64.b64encode(os.urandom(32)).decode()
# 基础配置类
class Config(object):
# 项目所在的绝对路径
BASE_DIR = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
# 加密密钥
SECRET_KEY = os.getenv('SECRET_KEY',
base64.b64encode(os.urandom(32)).decode())
class DevelopConfig(Config):
DEBUG = True
# sqlite数据库配置
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_DATABASE_URI = 'sqlite:////' + os.path.join(Config.BASE_DIR, 'sqlite.db')
class ProductConfig(Config):
DEBUG = False
# sqlite数据库配置
dev_db = 'mysql://root:mysql@localhost:3306/flask'
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URI', dev_db)
config = {
'development': DevelopConfig,
'production': ProductConfig,
'default': DevelopConfig
}创建APP
一个 Flask 应用是一个 Flask 类的实例。应用的所有东西（例如配置 和 URL ）都会和这个实例一起注册。
创建一个 Flask 应用最粗暴直接的方法是在代码的最开始创建一个全局 Flask 实例。有的情况下这 样做是简单和有效的，但是当项目越来越大的时候就会有些力不从心了。
可以在一个函数内部创建 Flask 实例来代替创建全局实例。这个函数被 称为 应用工厂 。所有应用相关的配置、注册和其他设置都会在函数内部完成， 然后返回这个应用。
app/__init__.py
from flask import Flask
from .config import config
from .extensions import config_extensions
from goods import goods
def create_app(config_name=None):
# 1. 创建APP
app = Flask(__name__)
# 2. 导入配置项
app.config.from_object(config.get(config_name) or config['default'])
# 3. 初始化插件
config_extensions(app)
# 4. 注册蓝图
app.register_blueprint(goods)
# 5. 返回APP
return app管理项目
如果是单独的app文件，直接运行即可。当项目庞大时，可以借助
flask_script来管理项目
manage.py
import os
from flask_script import Manager, Server
from flask_migrate import MigrateCommand
from app import create_app
config_name = os.getenv('FLASK_CONFIG', 'default')
app = create_app(config_name)
manage = Manager(app)
manage.add_command('db', MigrateCommand)
manage.add_command('serve', Server(host='0.0.0.0', port=7000))
if __name__ == '__main__':
manage.run()蓝图
在项目中创建一个包：
goods
创建蓝图
goods/__init__.py
from flask import blueprints
from flask_restful import Api
from .views import *
from .models import *
goods = blueprints.Blueprint('goods', __name__)
api = Api(goods)
# 后期可以在这对视图配置路由模型类
from app.extensions import db
class Cate(db.Model):
__tablename__ = 'tb_cate'
id = db.Column(db.Integer, autoincrement=True, primary_key=True)
name = db.Column(db.String(10))
# 外键，关联自身的主键
parent_id = db.Column(db.Integer, db.ForeignKey('tb_cate.id'), nullable=True)
# 指明关联字段，和实质的表无关系，只是在查询时，实现关联查询
parent = db.relationship('Cate', remote_side=[id], backref="sub")
def __repr__(self):
return 'Cate: %s' % self.name
class Goods(db.Model):
__tablename__ = 'tb_goods'
id = db.Column(db.Integer, autoincrement=True, primary_key=True)
name = db.Column(db.String(50))
price = db.Column(db.DECIMAL(10, 2))
stock = db.Column(db.Integer, default=0)
sales = db.Column(db.Integer, default=0)
# 外键，关联分类表的主键
cate_id = db.Column(db.Integer, db.ForeignKey('tb_cate.id'))
# 指明关联字段，和实质的表无关系，只是在查询时，实现关联查询
cate = db.relationship('Cate', backref='goods')
def __repr__(self):
return 'Goods: %s' % self.name已经在
manage.py中配置过模型类迁移命令：
python manage.py db init: 只有第一次才使用python manage.py db migrate: 每次模型类改变，就重新执行，生成迁移文件python manage.py db upgrate: 每次重新生成迁移文件，就重新执行，修改对应的数据库表
序列化字段
在序列化字段中，嵌套 列表、字典，不能直接嵌套，需要使用
List、Nested
goods/fields.py
from flask_restful import fields
cate_fields = {
'id': fields.Integer,
'name': fields.String,
'parent': fields.String(attribute='parent.name'),
}
cate2_fields = {
'id': fields.Integer,
'name': fields.String,
'sub': fields.List(fields.Nested(cate_fields))
}
cate1_fields = {
'id': fields.Integer,
'name': fields.String,
'sub': fields.List(fields.Nested(cate2_fields))
}
goods_fields = {
'id': fields.Integer,
'name': fields.String,
'stock': fields.Integer,
'sales': fields.Integer,
'cate': fields.Nested({
'id': fields.Integer,
'name': fields.String
}),
}
goods_page_fields = {
'data': fields.Nested(goods_fields),
'total': fields.Integer,
'pages': fields.Integer
}视图
goods/views.py
分类
列表、增加
from flask_restful import Resource, reqparse, marshal_with
from .models import *
from .ser import *
class CateView(Resource):
parser = reqparse.RequestParser()
@marshal_with(cate_fields)
def post(self):
# 1. 解析校验参数
self.parser.add_argument('name', type=str, required=True, location=['form', 'json'])
self.parser.add_argument('parent_id', type=int, location=['form', 'json'])
# 2. 获取参数
args = self.parser.parse_args()
# 2.1 校验父类是否存在
parent_id = args.get('parent_id')
if parent_id:
Cate.query.get_or_404(parent_id)
# 3. 添加数据
cate = Cate(**args)
# 4. 添加事务
db.session.add(cate)
# 5. 提交事务
db.session.commit()
# 6. 按照序列化字段解析分类对象，并返回响应
return cate, 201
# 实现三级分类解析
@marshal_with(cate1_fields)
def get(self):
cates = Cate.query.filter(Cate.parent_id.is_(None)).all()
return cates, 200详情、删除、修改
from flask_restful import Resource, reqparse, marshal_with
from .models import *
from .ser import *
class CateDetailView(Resource):
parser = reqparse.RequestParser()
@marshal_with(cate_fields)
def get(self, pk):
cate = Cate.query.get_or_404(pk)
return cate, 200
def delete(self, pk):
cate = Cate.query.get_or_404(pk)
db.session.delete(cate)
db.session.commit()
return None, 204
@marshal_with(cate_fields)
def put(self, pk):
cate = Cate.query.get_or_404(pk)
self.parser.add_argument('name', type=str, location=['form', 'json'], default=cate.name)
self.parser.add_argument('parent_id', type=int, location=['form', 'json'], default=cate.parent_id)
name = self.parser.parse_args().get('name')
parent_id = self.parser.parse_args().get('parent_id')
cate.name = name
cate.parent_id = parent_id
db.session.commit()
return cate, 201商品
列表分页、增加
from flask_restful import Resource, reqparse, marshal_with
from .models import *
from .ser import *
class GoodsView(Resource):
parser = reqparse.RequestParser()
@marshal_with(goods_fields)
def post(self):
self.parser.add_argument('name', type=str, location=['form', 'json'], required=True)
self.parser.add_argument('price', type=float, location=['form', 'json'], required=True)
self.parser.add_argument('stock', type=int, location=['form', 'json'], default=1000)
self.parser.add_argument('sales', type=int, location=['form', 'json'], default=0)
self.parser.add_argument('cate_id', type=int, location=['form', 'json'], required=True)
args = self.parser.parse_args()
goods = Goods(**args)
db.session.add(goods)
db.session.commit()
return goods, 201
@marshal_with(goods_page_fields)
def get(self):
self.parser.add_argument('cate_id', type=int, location='args')
self.parser.add_argument('page', type=int, default=1, location='args')
self.parser.add_argument('size', type=int, default=2, location='args')
args = self.parser.parse_args()
cate_id = args.get('cate_id')
page = args.get('page')
size = args.get('size')
pagination = Goods.query.filter_by(cate_id=cate_id).paginate(page, size, error_out=False) # 分页实现
# 注意：返回的数据结构需要和序列化字段结构一样
return {
"data": pagination.items, # 当前页的数据
'total': pagination.total, # 总的记录数
'pages': pagination.pages # 分页后的总页码
}, 200详情、删除、修改
from flask_restful import Resource, reqparse, marshal_with
from .models import *
from .ser import *
class GoodsDetailView(Resource):
parser = reqparse.RequestParser()
@marshal_with(goods_fields)
def get(self, pk):
goods = Goods.query.get_or_404(pk)
return goods, 200
def delete(self, pk):
goods = Goods.query.get_or_404(pk)
db.session.delete(goods)
db.session.commit()
return None, 204
@marshal_with(goods_fields)
def put(self, pk):
goods = Goods.query.get_or_404(pk)
self.parser.add_argument('name', type=str, location=['form', 'json'], default=goods.name)
self.parser.add_argument('price', type=float, location=['form', 'json'], default=goods.price)
self.parser.add_argument('stock', type=int, location=['form', 'json'], default=goods.stock)
self.parser.add_argument('sales', type=int, location=['form', 'json'], default=goods.sales)
self.parser.add_argument('cate_id', type=int, location=['form', 'json'], default=goods.cate_id)
name = self.parser.parse_args().get('name')
price = self.parser.parse_args().get('price')
stock = self.parser.parse_args().get('stock')
sales = self.parser.parse_args().get('sales')
cate_id = self.parser.parse_args().get('cate_id')
cate = Cate.query.get_or_404(cate_id)
goods.name = name
goods.cate_id = cate_id
goods.price = price
goods.stock = stock
goods.sales = sales
db.session.commit()
return goods, 201路由
goods/__init__.py
from flask import blueprints
from flask_restful import Api
from .views import *
from .models import *
goods = blueprints.Blueprint('goods', __name__)
api = Api(goods)
api.add_resource(CateView, '/cate/')
api.add_resource(CateDetailView, '/cate/<int:pk>/')
api.add_resource(GoodsView, '/goods/')
api.add_resource(GoodsDetailView, '/goods/<int:pk>/')
api.add_resource(GoodsCateChangeView, '/goods/<int:pk>/cate/')