利用Flask搭建API应用服务器

应用场景

      以电子商场为例，不同的电子商铺，货源完全有可能来源于同一个厂家。店铺只完成销售和付款，发货和物流完由厂家完成。这部分厂家的工作内容，相对于电子商城来说，属于应用服务器。

应用服务器和web服务器的不同点

        完整的web商城要同时解决，商城展示和数据管理两部分工作。而应用服务器只完成数据的处理和管理，而前端展示完全交给商家去做。这样，就很容易保持商品数据一致性的情况下。实现，用不同的方式在不同的店铺进行销售。

服务器文件结构

├─apps
   ├─api # 供应链物流API
├─db 数据库
├─lib 自定义插件

程序介绍

主程序  main.py

from apps import create_app
 
app = create_app()
 
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8360)

主程序初始化

 利用 flask API 创建 应用服务器，必须启动如下服务。

•  数据库服务
•  登录认证服务
•  API端口服务
•  跨域范围服务（第三方授权商城访问）

   这些服务必须跟随主程序启动时，一起激活。并且一个主程序，对同一类服务不能重复激活。

# 1.1 引入FLask 框架主函数
import click
from flask import Flask
 
# 1.2 引入数据库（数据库不能共搜）
from db.external import db
import db.external as external
 
# 1.3 引入登录验证机制和配置参数
from flask_migrate import Migrate
from lib.decorators import login_manager
from . import config
 
# 1.4 设置跨网站访问机制CORS
from flask_cors import CORS
 
 
# 创建主程序
def create_app():
    web = Flask(__name__, static_folder="../static", template_folder="../templates", static_url_path='/', )
 
    web.config.from_object(config)  # 导入配置文件
    web.config.from_object(external)
    register_extensions(web) # 初始化对象
    register_blueprints(web)  # API蓝本
    CORS(web, resources={r"/*": {"origins": "*"}} ) # sh
    return web
 
 
def register_extensions(app):
    db.init_app(app)
    Migrate(app,db)
    login_manager.init_app(app)
 
 
def register_blueprints(app):
    from apps.api import API_V1
    app.register_blueprint(API_V1, url_prefix='/' + app.config["API_POST"])  # 主视图函数蓝本
 
 
def register_commands(app):
    pass

数据库服务 external.py

#encoding:utf-8
from flask_sqlalchemy import SQLAlchemy #导入SQLAlchemy模块
 
#此时先不传入app 数据库初始化
db = SQLAlchemy()
 
HOST = '你自己的数据库地址'
PORT = '你自己的数据库端口'
USERNAME = 'root'
PASSWORD = '******'
# DATABASE = 'mysql'
DATABASE = 'apex_shop'
DB_URI = 'mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset-utf8'.format(username=USERNAME,
                                                                                       password=PASSWORD,
                                                                                        host=HOST,
                                                                                        port=PORT,
                                                                                   db=DATABASE)
 
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False
 
 
FLASKY_DB_QUERY_TIMEOUT=0.00005 #数据库查询时间的门限值

 系统配置文件 config.py

from datetime import timedelta
import random
 
# 这个是FLASK服务端的秘钥
SECRET_KEY = str(random.randint(100000, 999999))
 
EXPIRY_HOURS = 15
REFRESH_DAYS = 14
 
API_POST = 'api'

 API 端口服务 

flask 对于页面数据或应用数据，提供了蓝图模式。蓝图相当于仓库，不同的仓库可以管理不同的API应用。这里我们把对外API单独定为一个蓝图。

def register_blueprints(app):
    from apps.api import API_V1
    app.register_blueprint(API_V1, url_prefix='/' + app.config["API_POST"])  # 主视图函数蓝本

 flask 默认的API路由写法 是

@app.route('/hello',method=["GET","POST"])
def hello():
    return .....

  这种写法比较简单，但是在判断不同请求方式（GET、POST）撰写比较麻烦。

 这里我们采用另一种 flask_restful 的抒写方式。这种方式是把每个路由定义成一个类，列下面有get、post函数 来区不同的访问方法。

下面是写法参考。

class AuthGroup(Resource):
    def __init__(self):
        pass
 
    def post(self,methods):
 
        if methods == 'get.auth.group.list':
            role = Role.query.all()
            items = []
            for item in role:
                items.append({
                    "group_id": item.id,
                    "name": item.name,
                    "description": item.description,
                    "system": 1,
                    "sort": 50,
                    "status": 1
                })
            return jsonify(items)

 flask 登录验证

 flask 最成熟的登录模块是 flask_login，它支持 用户名密码登录和TOKEN登录两种方式。

 这里适合采用，TOKEN方式。下面简述一下。TOKEN登录的原理。详解在后续文章中描述。

from functools import wraps
from flask import session,redirect,url_for,request,jsonify,Markup,flash,abort,current_app,g
from db.models import User,Guest
from flask_login import LoginManager,login_user,logout_user,login_required,current_user,UserMixin
 
 
from apps.model import Token
import lib.base as base
import json
import apps.redis as rd
import time
 
#访问控制
# 实例化LoginManager类
login_manager = LoginManager()
 
# 提示信息
login_manager.login_message = "请先登录"
# 提示样式
login_manager.login_message_category = 'danger'
 
login_manager.anonymous_user = Guest
 
class bcolors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'
 
def load_token(tok):
    """通过loads()方法来解析浏览器发送过来的token，从而进行初步的验证"""
    try:
        api_key = Token.validate_token(tok)
        user = User.query.get(api_key.get('user_id'))
    except Exception as e:
        return None
 
    if user:
        user.token = tok
 
    else:
        print('用户不存在,令牌不正确!')
        return None
    return user
 
 
def get_token_key(key):
    """解密token到数据"""
    try:
        key = key.replace('Basic ', '', 1)
        key = key.encode('utf-8')
        key = Token.validate_token(key)
 
    except TypeError:
        return None
    return key
 
 
def get_key(value):
    if value == '':
        value = None
    return value
 
 
@login_manager.unauthorized_handler
def unauthorized():
    """系统超时退出"""
    return jsonify(base.get_result(401,'系统退出2'))
 
 
@login_manager.request_loader
def load_user_from_request(request):
    """效验请求中的token"""
    if request.method == "POST":
 
        g.refresh = False
        """# 请求正文中携带token时，执行如下代码
          捕捉请求参数中的 'api_key' 变量,这个api_key和数据库中存储的api_key比较，确认登录身份。 
        """
        api_access_key = request.args.get('api_key')
        if api_access_key:
            user = User.query.filter_by(api_key=api_access_key).first()
            if user:
                return user
 
        """请求头中携带token,执行如下代码
           捕捉请求参数中的 'X-Token' 变量,这个Access-Token通过服务端解码后，获得用户ID和Access-Token有效期，确认登录身份。
        """
        # 取请求头中的access-token
        api_access_key = get_key(request.headers.get('Access-Token'))
        api_refresh_key = get_key(request.headers.get('Refresh-Token'))
 
 
        # 对比前端的token和服务器中保存的token是否一致，判断用户是否已经登录了
        if api_access_key:
 
            login_data = get_token_key(api_access_key)
            Refresh_data = get_token_key(api_refresh_key)
 
            # 判断token是否过期，没有过期返回用户对象。
            if login_data and api_access_key == get_key(
                    rd.get_redis_data('front_access_token' + '_' + str(login_data.get('user_id')))):
                g.user = User.query.get(login_data.get('user_id'))
                return g.user
 
            # 判断如果token已经过期了，但是front_refresh_token没有过期，执行如下代码？
            elif Refresh_data:
                login_data = get_token_key(get_key(request.headers.get('Refresh-Token')))
 
                refresh_key = get_token_key(
                    rd.get_redis_data('front_refresh_token' + '_' + str(login_data.get('user_id'))))
                g.user = User.query.get(refresh_key.get('user_id'))
 
                g.refresh = True
                return g.user
 
            return None
 
        return None
 

启动程序

在CMD 中输入  flask run

这是通过浏览器访问 http:// 自己的服务器地址:自定义端口。就可以方法这个服务器获取数据了