Flask編寫api接口簡要流程 | 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