Flask编写api接口简要流程 | Flask

本文转载自查看原文 2021-12-18 00:18 94 Python/ Flask

使用Flask编写接口

Flask-Restful扩展的使用

项目文件：链接: https://pan.baidu.com/s/181V87QgrAtC7dCZb6fcBsQ 提取码: 7n22

首先需要用该插件编写请求过程中的参数获取->反序列化校验->逻辑处理->序列化响应

reqparse.RequestParser()来反序列化、校验参数

必须参数
```
required=True
```
前端请求的相同参数值放入列表
```
action='append'
```
前端请求参数与后端参数不同，对参数名称进行对应修改
```
dest='public_name'
```

从请求中不同位置获取参数，如：请求头、cookie、form请求体、files、args链接中参数

# location可以是个列表，从多个位置获取值
parser.add_argument('name', type=int, location='form')
parser.add_argument('limit', type=int, location='args')
parser.add_argument('User-Agent', location='headers')
parser.add_argument('session_id', location='cookies')
parser.add_argument('picture', type=werkzeug.datastructures.FileStorage, location='files')

对共享参数设置父解析器

parser_copy = parser.copy()  # 来共享
parser_copy.replace_argument('username', required=True, location='args') # 来修改父解析器的参数
parser_copy.remove_argument('age') # 来移除父解析器参数

捆绑错误处理

app.config['BUNDLE_ERRORS'] = True
# help设置错误信息，{error_msg}捕获异常错误输出
{
    "message": {
        "id": "error id.",
        "username": "error username.",
    }
}

marshal_with装饰器来序列化数据

响应数据字段重命名

resource_fields = {
    'user_name': fields.String(attribute='username'),  # attribute指定后端参数,user_name为前端展示字段名称
    'password': fields.String,
    'age':  fields.Integer,
}

指定默认值

resource_fields = {
    'user_name': fields.String(attribute='username', default='bob'),  # attribute指定后端参数,user_name为前端展示字段名称
    'password': fields.String,
    'age':  fields.Integer,
}

自定义类型与展示样式的映射关系

# 后端存储数字，前端展示对应值
class SexItem(fields.Raw):
    def format(self, value):
        c_map = {
            '1': '男',
            '0': '女',
        }

复杂字段

字段嵌套
```
fields.Nested({})
```
列表字段
```
fields.List(fields.String)
```

完整示例

# @file           : main.py
# @description    : flask主程序
# @date           : 2021/12/16 14:10:26
# @author         : miaokela
# @version        : 1.0
from flask import Flask
from flask_restful import Resource, Api, reqparse, fields, marshal_with, marshal_with_field

app = Flask(__name__)
app.config['BUNDLE_ERRORS'] = True  # 捆绑输出错误
app.config['RESTFUL_JSON'] = dict(ensure_ascii=False)  # 解决返回中文为unicode问题

# 将组件注册进flask应用
api = Api(app)

data = {
    '1': "hello",
    '2': "world",
}

class SexItem(fields.Raw):
    """
    自定义映射关系
    """
    def format(self, value):
        c_map = {
            '1': '男',
            '0': '女',
        }
        return c_map.get(value, '')


resource_fields = {
    'user_name': fields.String(attribute='username', default='bob'),  # attribute指定后端参数,user_name为前端展示字段名称
    'password': fields.String,
    'age':  fields.Integer,
    'sex': SexItem(attribute='person_sex'),
    'extra': fields.Nested({
        'address': fields.String,
        'phone': fields.String,
    }),
    'children': fields.List(fields.String)
}


class TestDao(object):
    """
    序列化输出类
    """
    def __init__(self, username, password, age, person_sex, extra, children):
        self.username = username
        self.password = password
        self.age = age
        self.person_sex = person_sex
        self.extra = extra
        self.children = children

class RestApiView(Resource):
    """
    请求视图类
    """
    
    def get(self, pk):
        # json值，状态码，请求头
        return {'pk': data[pk]}, 301, {'Referer': 'http://www.baidu.com'}

    @marshal_with_field(fields.List(fields.Integer))
    def post(self, pk):
        # 单个字段序列化返回
        """
        响应结果
            [1, 2, 3]
        """
        return ['1', 2, 3.0]

    @marshal_with(resource_fields, envelope='data')  # envelope对数据包裹一层
    def put(self, pk):
        """
        return c_map.get(value, '')
        :param <>:
        :return:
        """
        parser = reqparse.RequestParser()
        parser.add_argument('id', type=int, help='parse id error:{error_msg}')
        args = parser.parse_args(strict=True)

        c_id = args.get('id', '')
        print(c_id)

        # 后端查询数据
        test_data = {
            'username': 'miaokela',
            'password': '123456',
            'age': '10',
            'person_sex': '0',
            'extra': {
                'address': '大连',
                'phone': '1888888888'
            },
            'children': ['小明', '小安']
        }
        """
        响应结果:
        {
            "data": {
                "user_name": "miaokela",
                "password": "123456",
                "age": 10,
                "sex": "女",
                "extra": {
                "address": "大连",
                "phone": "1888888888"
                },
                "children": [
                "小明",
                "小安"
                ]
            }
        }
        """
        return TestDao(**test_data)

api.add_resource(RestApiView, '/<string:pk>')


if __name__ == '__main__':
    app.run()

项目搭建

Blueprint拆分视图

分发: views/user.py

from flask_restful import (
    Resource
)


class TestUser(Resource):

    def get(self):
        return {'data': 'user'}

分发: views/group.py

from flask_restful import (
    Resource
)


class TestGroup(Resource):

    def get(self):
        return {'data': 'group'}

BluePrint实现视图拆分: views/init.py

from flask import Blueprint


bp_api = Blueprint('api', __name__)

主程序注册蓝图: main.py

from flask import Flask
from flask_restful import Api
from views import (
    bp_api, user, group
)
from config import config

app = Flask(__name__)

app.config['BUNDLE_ERRORS'] = True  # 捆绑输出错误
app.config['RESTFUL_JSON'] = dict(ensure_ascii=False)  # 解决返回中文为unicode问题

# 将组件注册进flask应用
api = Api(bp_api)

api.add_resource(user.RestApiView, '/<string:pk>')
api.add_resource(group.TestGroup, '/group')

app.register_blueprint(bp_api, url_prefix='/api/v1.0')  # 注册蓝图，并且设置路由前缀


if __name__ == '__main__':
    app.run()

拆分出配置文件

config.py

class Config(object):
    """
    通用参数配置
    """
    BUNDLE_ERRORS = True # 捆绑输出错误
    RESTFUL_JSON = dict(ensure_ascii=False)  # 解决返回中文为unicode问题


class DevelopmentConfig(Config):
    """
    开发环境配置
    """
    DEBUG = True


class ProductionConfig(Config):
    """
    生产环境配置
    """
    pass


config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
}

读取配置文件

from flask import Flask
from flask_restful import Api
from views import (
    bp_api, user, group
)
from config import config

app = Flask(__name__)

# 将配置通过配置文件实现
app.config.from_object(config['development'])

# 将组件注册进flask应用
api = Api(bp_api)

api.add_resource(user.RestApiView, '/<string:pk>')
api.add_resource(group.TestGroup, '/group')

app.register_blueprint(bp_api, url_prefix='/api/v1.0')


if __name__ == '__main__':
    app.run()

通过命令行的方式启动应用

将main.py中操作应用app对象写入__init__.py

import Flask
from flask_restful import Api
from flask_pro.views import (
    bp_api, user, group
)
from flask_pro.config import config


def create_app(config_name='development'):
    """
    创建应用
    :param <config>: 选择指定环境的配置
    :return: app应用对象
    """
    app = Flask(__name__)

    # 将配置通过配置文件实现
    app.config.from_object(config[config_name])

    # 将组件注册进flask应用
    api = Api(bp_api)

    api.add_resource(user.RestApiView, '/<string:pk>')
    api.add_resource(group.TestGroup, '/group')

    app.register_blueprint(bp_api, url_prefix='/api/v1.0')
    return app

创建启动入口文件manage.py，使用Flask-Script扩展

from flask_script import Manager
from flask_pro import create_app

app = create_app()

manager = Manager(app)


if __name__ == '__main__':
    manager.run()

运行应用: python manage.py runserver

配置日志

import logging
from logging.handlers import RotatingFileHandler


# 设置日志的记录等级
logging.basicConfig(level=logging.DEBUG)  # 调试debug级
# 创建日志记录器，指明日志保存的路径、每个日志文件的最大大小、保存的日志文件个数上限
file_log_handler = RotatingFileHandler("logs/log", maxBytes=1024*1024*100, backupCount=10)
# 创建日志记录的格式                 日志等级    输入日志信息的文件名 行数    日志信息
formatter = logging.Formatter('%(levelname)s %(filename)s:%(lineno)d %(message)s')
# 为刚创建的日志记录器设置日志记录格式
file_log_handler.setFormatter(formatter)
# 为全局的日志工具对象（应用程序实例app使用的）添加日后记录器
logging.getLogger().addHandler(file_log_handler)

调用方式

from flask import current_app

current_app.logger.info('测试.')

接入SQLAlchemy

添加数据库配置

class DevelopmentConfig(Config):
    """
    开发环境配置
    """
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:password@127.0.0.1:3306/datacenter"
    SQLALCHEMY_TRACK_MODIFICATIONS = False  # will be disabled by default in the future
    SQLALCHEMY_ECHO = True  # 终端打印sql语句

模型文件

from flask_pro import db
from datetime import datetime

class BaseModel(object):
    """
    模型基类，为每个模型补充创建时间与更新时间
    """
    pass
    # create_time = db.Column(db.DateTime, default=datetime.now)  # 记录的创建时间
    # update_time = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)  # 记录的更新时间


class ReportInfo(BaseModel, db.Model):
    """
    报表信息
    """
    __tablename__ = 'datacenter_reportinfo'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=True)
    default_value = db.Column(db.String(100), nullable=True)
    sort = db.Column(db.Integer, nullable=True)
    state = db.Column(db.String(20), nullable=True)
    report_model_id = db.Column(db.Integer, db.ForeignKey(''), nullable=True)

    def to_dict(self):
        """
        自定义响应结果字段信息
        """
        return {
            'id': self.id,
            'name': self.name,
            'default_value': self.default_value,
            'sort': self.sort,
            'state': self.state,
            'report_model_id': self.report_model_id,
        }


class ReportModel(BaseModel, db.Model):
    """
    报表模板
    """
    __tablename__ = 'datacenter_reportmodel'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=True)
    code = db.Column(db.String(50), nullable=True)
    report_type = db.Column(db.String(20), nullable=True)

注册SQLAlchemy组件

# __init__.py
from flask import Flask
from flask_restful import Api

from flask_pro.config import config
import logging
from logging.handlers import RotatingFileHandler
from flask_sqlalchemy import SQLAlchemy

# 创建数据库对象
db = SQLAlchemy()  # 必须在应用模型类之前实例化，这里放在视图前

from .views import (
    bp_api, user, group
)

# 设置日志的记录等级
logging.basicConfig(level=logging.DEBUG)  # 调试debug级
# 创建日志记录器，指明日志保存的路径、每个日志文件的最大大小、保存的日志文件个数上限
file_log_handler = RotatingFileHandler("logs/log", maxBytes=1024*1024*100, backupCount=10)
# 创建日志记录的格式                 日志等级    输入日志信息的文件名 行数    日志信息
formatter = logging.Formatter('%(levelname)s %(filename)s:%(lineno)d %(message)s')
# 为刚创建的日志记录器设置日志记录格式
file_log_handler.setFormatter(formatter)
# 为全局的日志工具对象（应用程序实例app使用的）添加日后记录器
logging.getLogger().addHandler(file_log_handler)


def create_app(config_name='development'):
    """
    创建应用
    :param <config>: 选择指定环境的配置
    :return: app应用对象
    """
    app = Flask(__name__)

    # 将配置通过配置文件实现
    app.config.from_object(config[config_name])

    # 注册数据库信息
    db.init_app(app)
    # RuntimeError('No application found. Either work inside a view function or push an application context. See http://flask-sqlalchemy.pocoo.org/contexts/.',)
    # 将组件注册进flask应用
    api = Api(bp_api)

    api.add_resource(user.RestApiView, '/<string:pk>')
    api.add_resource(group.TestGroup, '/group')

    app.register_blueprint(bp_api, url_prefix='/api/v1.0')
    return app

CSRF保护机制(跨站请求伪造)

设置模板文件路劲与静态文件路径

app = Flask(
    __name__,
    template_folder='template',
    static_folder='statics'
)

GET请求返回FlaskForm对象

class TestGroup(Resource):

    def get(self):
        form = FlaskForm()

        return render_template('index.html', form=form)

前端页面展示csrf_token

<form action="">
    {{form.csrf_token()}}
</form>

请求参数中携带csrf_token参数

POST http://127.0.0.1:5000/api/v1.0/1
Content-Type: application/x-www-form-urlencoded

id=2&csrf_token=ImY1ZWYzMmY5N2FkMTdlMTIxNTQ2YWY1MmNlOGVlZDI3ODlkZjUzYTYi.Ybyv6g.xQSsAKlAabPiY1Y7aTtxbyL4KaM

Gunigorn、Nginx部署Flask应用

Gunigorn实现WSGI协议

安装命令
```
pip install gunicorn    
```
运行命令
默认使用的是8000 可以通过-b 127.0.0.1:5000 设置到5000或其他端口
```
gunicorn -w 4 manage:app    
```

Nginx负责反向代理、请求转发、负载均衡、静态文件代理

修改配置文件(/usr/local/etc/nginx)

location / {
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}

常用命令

brew services start nginx  # macos
nginx -s stop
nginx -s reload

免责声明！

本站转载的文章为个人学习借鉴使用，本站对版权不负任何法律责任。如果侵犯了您的隐私权益，请联系本站邮箱yoyou2525@163.com删除。

猜您在找 python编写api接口--使用flask Flask 编写http接口api及接口自动化测试基于Flask的 api（一） Flask 框架：实现简单API测试接口 python flask api 接口开发（使用 Python 和 Flask 设计 RESTful API） Python+Flask之数据库接口的编写 python+flask编写一个简单的登录接口例子 Flask - 底层原理和基本流程 flask请求流程详解 Flask设计带认证token的RESTful API接口[翻译]