7、flask与flask-restful的结合使用

搭建一个完整的flask服务：

配置文件: config.py

# -*- coding:utf-8 -*-

# 生产环境
class Config(object):
    DEBUG = True
    # 数据库 参数设置
    SQLALCHEMY_ECHO = True
    SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://192.168.XXX.XXX/flask_db?user=root&password=******'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    # 连接池 设置
    SQLALCHEMY_POOL_SIZE = 10
    SQLALCHEMY_POOL_TIMEOUT = 10
    SQLALCHEMY_POOL_RECYCLE = 10

    @staticmethod
    def init_app(app):
        pass


# 测试环境
class TestConfig(object):
    DEBUG = True
    # 数据库 参数设置
    SQLALCHEMY_ECHO = True
    SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://192.168.XXX.XXX/flask_db?user=root&password=******'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    # 连接池 设置
    SQLALCHEMY_POOL_SIZE = 10
    SQLALCHEMY_POOL_TIMEOUT = 10
    SQLALCHEMY_POOL_RECYCLE = 10

    @staticmethod
    def init_app(app):
        pass


# 环境切换设置
config = {
    'production': Config,
    'testing': TestConfig
}

启动文件： proj_dir/crm/app.py

# -*- coding:utf-8 -*-
from crm import create_app

app = create_app(environment='testing')


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5001, debug=True)

配置启动信息： proj_dir/crm/__init__.py

# -*- coding:utf-8 -*-
from flask import Flask
from config import config
from crm.shared.db import db
from crm.shared.models import metadata

def create_app(environment='production'):
    app = Flask(__name__)
    # 添加配置
    app.config.from_object(config[environment])

    # 导入数据库基本设置，并加载设置
    db.init_app(app)
    app.app_context().push()

    # 获取引擎
    metadata.create_all(db.get_engine())

    # 蓝图
    from .api import api_blueprint
    app.register_blueprint(api_blueprint, url_prefix='/api')

    return app

数据库创建： proj_dir/crm/shared/db.py

# -*- coding:utf-8 -*-
from flask_sqlalchemy import SQLAlchemy

# 数据库
db = SQLAlchemy()

数据库映射： proj_dir/crm/shared/models.py

# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String
from sqlalchemy.dialects.mysql import INTEGER


# 声明数据库映射
Base = declarative_base()
metadata = Base.metadata


# 覆写
def to_dict(self):
    return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}


Base.to_dict = to_dict


class UserDict(Base):
    __tablename__ = 'user'

    user_id = Column(INTEGER(11), primary_key=True)
    user_name = Column(String(100), nullable=False)
    date = Column(String(100), nullable=False)
    address = Column(String(100), nullable=False)

实例化蓝图、加载资源： proj_dir/crm/api/__init__.py

# -*- coding:utf-8 -*-
from flask import Blueprint
from flask_restful import Api
from crm.api.user import User, UserList


def register_views(app):
    api = Api(app)
    # 资源
    api.add_resource(User, '/user')
    api.add_resource(UserList, '/users')


# 注册
api_blueprint = Blueprint('api', __name__)
register_views(api_blueprint)


# 导入视图函数模块 (初始化)
from . import user

数据解析、资源类创建： proj_dir/crm/api/user.py

# -*- coding:utf-8 -*-
from flask import jsonify
from flask_restful import Resource, reqparse
from crm.shared.db import db
from crm.shared.models import UserDict


# 解析用户输入的数据，尽量做到多个格式兼容
parse = reqparse.RequestParser()
parse.add_argument('user_id', type=int)
parse.add_argument('user_name', type=str)
parse.add_argument('date', type=str)
parse.add_argument('address', type=str)
parse.add_argument('limit', type=int)
parse.add_argument('skip', type=int)



class User(Resource):
    # 查询
    def get(self):
        args = parse.parse_args()
        user_name = args['user_name']
        # 查询数据库
        user_info = db.session.query(
            UserDict.user_id, UserDict.user_name, UserDict.date,
            UserDict.address).filter(UserDict.user_name == user_name).first()

        result = {
            'user_id':user_info[0],
            'user_name': user_info[1],
            'date': user_info[2],
            'address': user_info[3]
        }
        return jsonify(result)

    # 创建
    def post(self):
        args = parse.parse_args()
        date = args['date']
        user_name = args['user_name']
        address = args['address']
        # 数据转换
        user_info = UserDict(
            user_name=user_name,
            date=date,
            address=address
        )
        db.session.add(user_info)
        db.session.commit()

        return user_info.user_id, 201


class UserList(Resource):
    def get(self):
        args = parse.parse_args()
        limit = args['limit']
        skip = args['skip']

        user_filter = db.session.query(UserDict.user_id, UserDict.user_name,  UserDict.date, UserDict.address).filter()
        # 过滤
        if limit and skip:
            user_filter = user_filter.limit(limit).offset(skip)

        # 数据
        users = user_filter.all()
        total_count = user_filter.count()

        # 拼接结果
        result = {
            'data': list(
                map(lambda user_info: {
                    'user_id': user_info[0],
                    'user_name': user_info[1],
                    'date': user_info[2],
                    'address': user_info[3]
                }, users)
            ),
            'total_count': total_count
        }

        return jsonify(result)

Test：

# POST
http://0.0.0.0:5001/api/user
{
    "user_name": "张一鸣",
    "date": "2016-04-01",
    "address": "北京市"
}

# GET
http://0.0.0.0:5001/api/user
{
    "user_name": "Robbin"
}

# GET
http://0.0.0.0:5001/api/users
{
    "limit": 3,
    "skip": 2
}