使用 Flask + Flask RESTful 快速搭建 API 服务

简介

Flask 是 Python 社区中开发 Web 运用最炽热的结构之一，不同于 Django 峻峭的学习曲线，个人感觉 Flask 十分好上手，且社区生态丰厚，有很多成熟的扩展能够拿来直接装置运用。 Flask 结构自身集成了根据 Jinja 的模板语言，使其能够完结前后端的一切开发，但现在大部分的 Web 运用都是前后端别离，所以本文将运用 Flask RESTful 扩展完结一个纯后端的 API 服务。 经过本文能够学习到以下内容：

• 运用 Flask + Flask RESTful 建立 API 运用并运用 Blueprint(蓝图) 办理 API；
• 运用 Flask-SQLAlchemy 扩展完结 ORM 操作 MySQL 数据库；
• 根据 JWT 验证完结注册、登录以及登出接口；
• 完结一个最基本的列表获取接口；
• 处理跨域问题；
• 运用 Docker 布置该运用。

在正式开端之前，请保证你现已装置了 Python，并且对 Python 有必定了解。

初始化项目

首要咱们新建一个空文件夹，作为项目的根目录。进入到项目根目录后创立一个虚拟环境：

# MacOS/Linux
mkdir myproject
cd myproject
python3 -m venv .venv
​
# Windows
mkdir myproject
cd myproject
py -3 -m venv .venv

激活虚拟环境

# MacOS/Linux
. .venv/bin/activate
# Windows
.venv\Scripts\activate

装置 Flask、Flask RESTful 以及 python-dotenv，最终一个包用来获取咱们在项目中界说的环境变量。

pip install Flask flask-restful python-dotenv

依照常规，咱们先简略完结一个接口，验证下咱们最基础的包是否装置完结。 首要在项目根目录下新建一个 app 文件夹，在 app 下新建一个 __init__.py 文件，在这个文件中，咱们界说一个测试用的接口。

from flask import Flask
from flask_restful import Resource, Api
​
​
app = Flask(__name__)
api = Api(app)
​
class HelloWorld(Resource):
  def get(self):
    return {'hello': 'world'}
​
api.add_resource(HelloWorld, '/')

咱们先引进了 Flask 和 flask_restful 中的 Resource 和 Api，然后咱们运用 Flask() 初始化一个 Flask 运用实例赋值给 app，传入的 __name__ 则是模块名 "app"，然后再运用 Api(app) 初始化一个 flask_restful 实例赋值给 api。 接下来咱们界说了 HelloWorld 这个类，它承继于 Resource 类。这个类中界说一个名为 get 的函数，它返回一个固定的 JSON 为{'hello': 'world'}。

最终咱们运用 api.add_resource(HelloWorld, '/') 去注册接口，并指定了拜访路由，当拜访的接口途径为 "/" 且恳求办法为 GET 时，就会调用该类中界说好的 get() 函数处理。 你或许现已猜到了，在以 Resource 类为基类的派生类中，就是咱们界说不同 HTTP 恳求办法的当地，所以在这个类中，你还能够界说 post，put，delete 等函数。 接下来，咱们在项目根目录中新建一个 run.py 文件：

from app import app
​
​
if __name__ == '__main__':
  app.run(host="0.0.0.0", port=5000, debug=True)

这是一个发动文件，你能够直接在操控台中运用 python run.py 履行，或是和我相同，在项目根目录下新建一个 .env 文件用来寄存环境变量：

FLASK_ENV=development # 当时环境
FLASK_DEBUG=True # 开启 debug mode
FLASK_APP=run.py # flask项目进口文件

上面这些环境变量的命名办法都是 Flask 规定的，这样指定环境变量的优点就是咱们能够经过操控台履行 flask run 指令来发动服务。 需求留意的是，假如你经过 flask run 指令来发动服务，那么 Flask 的装备会默许以环境变量为准，并且会疏忽 run.py 中的装备项。 现在咱们发动项目后，看到以下信息就阐明服务发动成功了。

• Serving Flask app ‘run.py’
• Debug mode: on
• Running on http://127.0.0.1:5000 Press CTRL+C to quit
• Restarting with stat
• Debugger is active!
• Debugger PIN: xxx-xxx-xxx

现在你能够直接运用浏览器拜访 http://127.0.0.1:5000/ 或运用 Apifox 等接口调试东西来进行测试，看看是否会得到返回的 {'hello': 'world'} JSON 字符串。

衔接数据库

要衔接数据库，需求先装置 Flask-SQLAlchemy 、Flask-Migrate 和 pymysql 这三个扩展包，Flask-SQLAlchemy 用来衔接并操作数据库，Flask-Migrate 是一个数据库搬迁插件，用来同步数据库的更改。最终由于咱们要衔接的是 MySQL 数据库，所以需求装置一个 pymysql 包。

pip install Flask-SQLAlchemy Flask-Migrate pymysql

接下来咱们修正一下咱们项目的目录结构，让咱们项目的可扩展性更强。

/
├── .venv/
├── app/
│  └── api/ # api 接口模块
│    └── __init__.py # 注册以及生成蓝图
│    └── common/ # 公共办法
│    └── models/ # 模型
│    └── resources/ # 接口
│    └── schema/ # 校验
│  └── __init__.py # 整个运用的初始化
│  └── config.py # 装备项
│  └── manage.py # 数据库搬迁东西办理
├── .env # 环境变量
├── run.py # 进口文件

这样修正咱们的项目结构，能够让咱们合作 Flask 供给的 Blueprint （蓝图） 对接口进行模块化办理和开发，有助于提高咱们项目的可扩展性和可保护性。 接下来咱们先装备数据库衔接，首要你要保证你有可用的 MySQL 数据库，然后在 /app/config.py 中添加如下代码：

import os
​
# 数据库相关装备
# 主张在本地根目录下新建 .env 文件保护敏感信息装备项更安全 
# 用户名
USERNAME = os.getenv('MYSQL_USER_NAME')
# 暗码
PASSWORD = os.getenv('MYSQL_USER_PASSWORD')
# 地址
HOSTNAME = os.getenv('MYSQL_HOSTNAME')
# 端口
PORT = os.getenv('MYSQL_PORT')
# 数据库称号
DATABASE = os.getenv('MYSQL_DATABASE_NAME')
​
# 固定格式 不必改
DIALECT = 'mysql'
DRIVER = 'pymysql'
​
class Config(object):
  DEBUG = False
  TESTING = False
  SECRET_KEY = os.getenv('SECRET_KEY')
  SQLALCHEMY_DATABASE_URI = "{}+{}://{}:{}@{}:{}/{}?charset=utf8".format(DIALECT, DRIVER, USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
  SQLALCHEMY_ECHO = True
​
class ProductionConfig(Config):
  DEBUG = False
  SQLALCHEMY_DATABASE_URI = ''
​
class DevelopmentConfig(Config):
  DEBUG = True
​
class TestingConfig(Config):
  TESTING = True
​
config = {
  'development': DevelopmentConfig,
  'production': ProductionConfig,
  'testing': TestingConfig,
  'default': DevelopmentConfig,
}

然后咱们在 .env 文件中新增下面的环境变量：

# ...
SECRET_KEY=b'#q)\\x00\xd6\x9f<iBQ\xd7;,\xe2E' # flask扩展密钥
MYSQL_USER_NAME=<your mysql username> # MySQL数据库称号
MYSQL_USER_PASSWORD=<your mysql pwd> # MySQL数据库暗码
MYSQL_HOSTNAME=<your mysql hostname>
MYSQL_PORT=<your mysql port>
MYSQL_DATABASE_NAME=<your mysql database name>

留意： 将 <> 中的内容替换成你自己的装备。 在 /app/manage.py 文件中，初始化数据库搬迁东西：

from flask_migrate import Migrate
​
​
migrate = Migrate()

接下来咱们修正 /app/__init__.py 文件：

import os
​
from flask import Flask
​
from .config import config
from .api.models import db
from .api import api_blueprint
from .manage import migrate
​
​
def create_app(config_name):
  # 初始化 Flask 项目
  app = Flask(__name__)
  # 加载装备项
  app.config.from_object(config[config_name])
  # 初始化数据库ORM
  db.init_app(app)
  # 初始化数据库ORM搬迁插件
  migrate.init_app(app, db)
  # 注册蓝图
  app.register_blueprint(api_blueprint)
​
  return app
​
​
# 初始化项目
app = create_app(os.getenv('FLASK_ENV', 'development'))

现在还缺少一个 db 和一个 api_blueprint 变量，咱们先把 db 补上。 咱们在 /app/api/models 下新建一个 __init__.py 文件，在里面初始化 Flask-SQLAlchemy 扩展：

from flask_sqlalchemy import SQLAlchemy
​
​
db = SQLAlchemy()

运用蓝图

接下来咱们弥补 api_blueprint 。 在上一节的初始化中，咱们写了一个 HelloWorld 类并注册到 / 路由上，在实践开发中，咱们并不会这么做，而是将事务接口拆分模块，比如 /api/xxx，所以现在咱们需求创立一个 api 蓝图来统一办理，在 /app/api/__init__.py 文件中写入以下代码：

from flask import Blueprint
from flask_restful import Api
​
​
api_blueprint = Blueprint('api', __name__, url_prefix="/api")
api = Api(api_blueprint)

咱们先运用 Flask 中的 Blueprint 新建一个蓝图，将前缀设置为 /api，然后咱们运用 flask_restful 中的 Api() 初始化这个蓝图，假设咱们在该文件中运用 api.add_resource(HelloWorld, '/greet') 注册了接口，那么只需求经过 /api/greet 进行调用即可。

创立模型并更新数据库

这一节咱们创立 User 模型，并且运用 ORM 生成对应的数据库表，首要在 /app/api/models 下新建一个 user.py，咱们在这个文件中界说 User 表：

from ..models import db
from datetime import datetime
​
​
class UserModel(db.Model):
  """
   用户表
   """
  __tablename__ = 'user'
​
  # 主键 id
  id = db.Column(db.Integer(), primary_key=True, nullable=False, autoincrement=True, comment='主键ID')
  # 用户名
  username = db.Column(db.String(40), nullable=False, default='', comment='用户名字')
  # 暗码
  pwd = db.Column(db.String(102), comment='暗码')
  # salt
  salt = db.Column(db.String(32), comment='salt')
  # 创立时刻
  created_at = db.Column(db.DateTime(), nullable=False, default=datetime.now, comment='创立时刻')
  # 更新时刻
  updated_at = db.Column(db.DateTime(), nullable=False, default=datetime.now, onupdate=datetime.now, comment='更新时刻')
​
  # 新增用户
  def addUser(self):
    db.session.add(self)
    db.session.commit()
​
  # 用户字典
  def dict(self):
    return {
      "id": self.id,
      "username": self.username,
      "created_at": self.created_at,
      "updated_at": self.updated_at,
     }
  
  # 获取暗码和 salt
  def getPwd(self):
    return {
      "pwd": self.pwd,
      "salt": self.salt,
     }
​
  # 按 username 查找用户
  @classmethod
  def find_by_username(cls, username):
    return db.session.execute(db.select(cls).filter_by(username=username)).first()
​
   # 返回一切用户
  @classmethod
  def get_all_user(cls):
    return db.session.query(cls).all()

咱们界说了 User 表中一些必要的字段以及一些常用的办法，比如新增和查询。这些办法咱们后边会用到。接下来咱们需求在 /app/__init__.py 中显式的引进该模型：

# ...
from .manage import migrate
# *important 一切数据库模型 需求显式的在这里导入
from .api.models.user import UserModel
​
​
def create_app(config_name):
   # ...

现在咱们要在操控台履行数据库搬迁东西的同步指令，来查验数据库东西是否可用：

# 第一次初始化时运用
flask db init 
# 后边每次修正数据库字段时运用
flask db migrate
flask db upgrade

需求留意的是，flask db init 只在咱们新项目第一次初始化数据库时运用，后续有表字段修正以及新增表的时候，只需求履行后边两条指令即可。现在咱们翻开数据库东西查看，应该现已有两个表了。

其间 user 表是咱们刚创立的表，而 alembic_version 表是 Flask-Migrate 扩展主动创立的，翻开这个表，里面有且只要一个字段 version_num，该字段是记录你的数据库更新搬迁版本号的，这个表不要随便改动，让 Flask-Migrate 自行办理就好。

完结注册接口

接下来咱们完结注册接口，首要在 /app/api/resources 下创立 register.py 文件，开端写注册接口的逻辑：

import uuid
​
from flask_restful import Resource, reqparse
from werkzeug.security import generate_password_hash
​
from ..models.user import UserModel
​
​
class Register(Resource):
  def post(self):
    parser = reqparse.RequestParser()
    parser.add_argument('username', type=str, location='json')
    parser.add_argument('password', type=str, dest='pwd', location='json')
    data = parser.parse_args()
    if UserModel.find_by_username(data['username']):
      return {
        'success': False,
        'message': "Repeated username!",
        'data': None,
       }, 400
    else: 
      try:
        data['salt'] = uuid.uuid4().hex
        data['pwd'] = generate_password_hash('{}{}'.format(data['salt'], data['pwd']))
        user = UserModel(**data)
        user.addUser()
        return {
          'success': True,
          'message': "Register succeed!",
          'data': None,
         }, 200
      except Exception as e:
        return {
          'success': False,
          'message': "Error: {}".format(e),
          'data': None,
         }, 500

首要咱们界说了一个承继自 Resource 类的 Register 派生类，在类里面咱们界说了一个 post 函数，然后运用 reqparse 界说了接口的参数校验包括 username 和 password 两个字段，类型都是 string，取参数的方位是 json，dest 则表示设置了参数的别名，解析参数之后只需求用 pwd 即可取到恳求传来的 password 参数。

接着咱们判别了传过来的用户名是否重复，假如重复了则抛出过错信息，假如没有重复，咱们将用户的暗码进行 MD5 + Salt 加密，最终咱们在数据库里储存加密之后的暗码和 Salt。一起咱们对整个加密的进程进行过错捕获，以防程序履行时报错无法通知到客户端。 接下来咱们在 /app/api/__init__.py 中去注册这个接口：

# ...
from .resources.register import Register
​
# ...
api = Api(api_blueprint)
​
api.add_resource(Register, '/register')

Flask 默许会在开发形式下开启热更新，检测到你代码修正后它会重启服务，所以咱们无需重启服务，能够直接运用调试东西进行接口测试。 再进行下一步之前，咱们先优化下代码，咱们上面注册接口的代码其实是有优化空间的，能够将不重要的参数校验以及重复性的 Response 内容抽离封装。 咱们先抽离参数校验部分，在 /app/api/schema 下新建一个 register_sha.py 文件，把参数校验逻辑转移到该文件内：

def reg_args_valid(parser):
  parser.add_argument('username', type=str, location='json')
  parser.add_argument('password', type=str, dest='pwd', location='json')

然后咱们在 /app/api/common 下新建一个 utils.py 文件，封装一个公共的 Response 办法：

# 公共 response 办法
def res(data=None, message='Ok', success=True, code=200):
  return {
    'success': success,
    'message': message,
    'data': data,
   }, code

最终咱们修正一下 /app/api/resources/register.py 文件：

# ...
from werkzeug.security import generate_password_hash
​
from ..common.utils import res
from ..models.user import UserModel
from ..schema.register_sha import reg_args_valid
​
​
class Register(Resource):
  def post(self):
    parser = reqparse.RequestParser()
    reg_args_valid(parser)
    data = parser.parse_args()
    if UserModel.find_by_username(data['username']):
      return res(success=False, message="Repeated username!", code=400)
    else: 
      try:
        data['salt'] = uuid.uuid4().hex
        data['pwd'] = generate_password_hash('{}{}'.format(data['salt'], data['pwd']))
        user = UserModel(**data)
        user.addUser()
        return res(message="Register succeed!")
      except Exception as e:
        return res(success=False, message="Error: {}".format(e), code=500)

完结登录接口

现在咱们来完结登录接口，在开端之前，咱们需求装置 Flask-JWT-Extended 扩展来协助咱们完结 Token 的创立以及校验等工作。

pip install Flask-JWT-Extended

装置完结后咱们在 /app/config.py 中添加以下内容：

# ...
from datetime import timedelta
# ...
class Config(object):
  # ...
  JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY')
  JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=2)
  JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
  JWT_BLOCKLIST_TOKEN_CHECKS = ['access']
# ...

在 /.env 中添加环境变量：

# ...
JWT_SECRET_KEY=b'#q)\\x00\xd6\x9f<iBQ\xd7;,\xe2E' # jwt密钥
# ...

在 /app/__init__.py 中初始化 JWT 扩展：

# ...
from flask_jwt_extended import JWTManager
# ...
def create_app(config_name):
  # ...
  # 初始化 JWT
  jwt = JWTManager(app)
  return app
# ...

初始化扩展后，咱们在 /app/api/resources 下新建 login.py，并完结登录接口的逻辑：

from flask_restful import Resource, reqparse
from flask_jwt_extended import create_access_token, create_refresh_token
from werkzeug.security import check_password_hash
​
from ..schema.register_sha import reg_args_valid
from ..models.user import UserModel
from ..common.utils import res
​
​
class Login(Resource):
  def post(self):
    # 初始化解析器
    parser = reqparse.RequestParser()
    # 添加恳求参数校验
    reg_args_valid(parser)
    data = parser.parse_args()
    username = data['username']
    user_tuple = UserModel.find_by_username(username)
    if user_tuple:
      try:
         (user,) = user_tuple
        pwd, salt  = user.getPwd().get('pwd'), user.getPwd().get('salt')
        valid = check_password_hash(pwd, '{}{}'.format(salt, data['pwd']))
        if valid:
          # 生成 token 
          response_data = generateToken(username)
          return res(response_data)
        else:
          raise ValueError('Invalid password!')
      except Exception as e:
        return res(success=False, message='Error: {}'.format(e), code=500)
    else:
      return res(success=False, message='Unregistered username!', code=400)
​
# 生成token
def generateToken(id):
  access_token = create_access_token(identity=id)
  refresh_token = create_refresh_token(identity=id)
  return {
    'accessToken': 'Bearer ' + access_token,
    'refreshToken': 'Bearer ' + refresh_token,
   }

相同的，咱们新建了一个 Login 类，并且界说了一个 post 函数表明该接口是 POST 恳求。由于登录接口传入的参数和注册接口一致，所以直接引进注册接口的校验函数。解析完参数后，判别该用户是否现已注册，假如没注册则抛出过错，假如注册了则进行暗码校验，校验经过了就运用扩展供给的函数新建两个 Token，其间 access_token 是用来鉴权的，有效期 2 小时（在 config.py 中装备的），而为了防止用户需求频频的重新登录，再生成一个refresh_token，当access_token 过期后运用 refresh_token 来交换新的 access_token，当然，refresh_token 也有 30 天的有效期。 接下来咱们再写一下交换 Token 的接口：

from flask_restful import Resource, reqparse
from flask_jwt_extended import create_access_token, create_refresh_token, jwt_required, get_jwt_identity
# ...
class Login(Resource):
  def post(self):
    # ...
​
  @jwt_required(refresh=True)
  def get(self):
    # access_token 过期后 需求用 refresh_token 来交换新的 token
    # 先从 refresh_token 中取出用户信息
    current_username = get_jwt_identity()
    # 再生成新的 token
    access_token = create_access_token(identity=current_username)
    return res(data={'accessToken': 'Bearer ' + access_token})
​
# 生成token
def generateToken(id):
  access_token = create_access_token(identity=id)
  refresh_token = create_refresh_token(identity=id)
  return {
    'accessToken': 'Bearer ' + access_token,
    'refreshToken': 'Bearer ' + refresh_token,
   }

咱们直接在 Login 类中在声明一个 get 函数，然后加上 @jwt_required 装饰器，当加上该装饰器时，JWT 扩展会为咱们主动在调用此接口时做 Token 校验，它默许是只校验 access_token 的，在括号内传入 refresh=True 则表示用有效的 refresh_token 也能够经过校验。 接下来咱们在 /app/api/__init__.py 中注册该接口：

# ...
from .resources.register import Register
from .resources.login import Login
# ...
api.add_resource(Register, '/register')
api.add_resource(Login, '/login', '/refreshToken')

完结登出接口

在用户退出登录后，咱们要毁掉 Token，接下来咱们来完结这个接口。首要咱们需求一个表来寄存现已毁掉的 Token，在 /app/api/models 下新建 revoked_token.py 文件：

from ..models import db
​
​
class RevokedTokenModel(db.Model):
  """
     已过期的token表
   """
​
  __tablename__ = 'revoked_tokens'
​
  # 主键 id
  id = db.Column(db.Integer, primary_key=True)
  # jwt 仅有标识
  jti = db.Column(db.String(120))
​
  # token 加黑
  def add(self):
    db.session.add(self)
    db.session.commit()
​
  # 查询是否是加黑的 token
  @classmethod
  def is_jti_blacklisted(cls, jti):
    query = cls.query.filter_by(jti=jti).first()
    return bool(query)

咱们创立一个 revoked_tokens 表，用来寄存现已毁掉的 Token，并且界说一个查询的办法，用来查询 Token 是否已毁掉。 然后咱们在 /app/api/resources 下新建 logout.py 写入登出接口逻辑：

from flask_restful import Resource
from flask_jwt_extended import jwt_required, get_jwt
from ..models.revoked_token import RevokedTokenModel
from ..common.utils import res
​
class Logout(Resource):
  @jwt_required()
  def post(self):
    jti = get_jwt()['jti']
    try:
      # 用户退出体系时 将 token 加入黑名单
      revoked_token = RevokedTokenModel(jti=jti)
      revoked_token.add()
      return res()
    except:
      return res(success=False, message='服务器繁忙！', code=500)

在用户退出登录时，咱们先获取到 Token 中的仅有标识 jti 然后将它加入毁掉 Token 的表中。 现在咱们在 /app/api/__init__.py 去注册该接口：

# ...
from .resources.logout import Logout
# ...
api.add_resource(Logout, '/logout',)

接下来咱们需求注册一个 JWT 扩展供给的钩子函数，用来校验 Token 是否在毁掉列表中。在 /app/__init__.py 中添加以下内容：

# ...
# *important 一切数据库模型 需求显式的在这里导入
from .api.models.user import UserModel
from .api.models.revoked_token import RevokedTokenModel
​
​
def create_app(config_name):
# ...
  # 初始化 JWT
  jwt = JWTManager(app)
  # 注册 JWT 钩子
  registerJwtHooks(jwt)
  return app
​
def registerJwtHooks(jwt):
  # 注册 JWT 钩子 用于查看 token 是否在黑名单中
  @jwt.token_in_blocklist_loader
  def check_if_token_in_blacklist(jwt_header, decrypted_token):
    jti = decrypted_token['jti']
    return RevokedTokenModel.is_jti_blacklisted(jti)
# ...

至此，当用户在调用咱们需求鉴权的接口时，JWT 扩展还会帮咱们校验是否是现已毁掉的 Token。 由于咱们方才新增了一张表，所以需求履行下数据库搬迁扩展的更新指令再重启服务。

flask db migrate
flask db upgrade

完结获取用户列表接口

最终咱们完结一个获取用户列表的接口，刚好能够测试咱们的鉴权逻辑是否都完结了，在 /app/api/resources 下新建 user.py 文件：

from flask_restful import Resource
from flask_jwt_extended import jwt_required
from ..models.user import UserModel
from ..common.utils import res
​
class UserService(Resource):
  @jwt_required()
  def get(self):
    userList = UserModel.get_all_user()
    result = []
    for user in userList:
      result.append(user.dict())
      
    return res(data=result)

咱们界说了一个 UserService 类，在类中界说了一个 get 函数，并且添加了 @jwt_required() 装饰器，表示该接口需求鉴权，调用该接口返回 user 表中一切的用户。 在 /app/api/__init__.py 中注册该接口：

# ...
from .resources.user import UserService
# ...
api.add_resource(UserService, '/getUserList')

咱们的 user.dict() 办法中，返回了两个时刻字段，由于 Python 中 datetime 格式不能直接放在 JSON 中返回，所以现在咱们需求新写一个将 datetime 转化格式的办法，在 /app/api/common/utils.py 中新增：

# 公共 response 办法
def res(data=None, message='Ok', success=True, code=200):
# ...
​
# datetime 转化格式
def format_datetime_to_json(datetime, format='%Y-%m-%d %H:%M:%S'):
  return datetime.strftime(format)

修正 /app/api/models/user.py 中 dict() 函数：

#...
from ..common.utils import format_datetime_to_json
​
​
class UserModel(db.Model):
  """
   用户表
   """
  # ...
  # 用户字典
  def dict(self):
    return {
      "id": self.id,
      "username": self.username,
      "created_at": format_datetime_to_json(self.created_at),
      "updated_at": format_datetime_to_json(self.updated_at),
     }
  # ...

处理跨域问题

现在咱们完结了一切接口的开发，假如需求在浏览器环境中进行接口调用，但前后端服务又不同源的情况下，是会呈现跨域问题的，所以咱们需求装置 Flask-Cors 扩展来处理这个问题。

pip install Flask-Cors

在 /app/__init__.py 中运用该扩展：

# ...
from flask_cors import CORS
#...
def create_app(config_name):
  # ...
  # 处理跨域
  CORS(app)
  # ...

至此，咱们经过最简略的设置 Access-Control-Allow-Origin: * 响应头来操控资源跨域同享。

经过 Docker 布置

最终咱们运用 Docker 将项目布置在服务器上，我这里服务器体系运用的 Linux 发行版是 Ubuntu 18.04，在布置之前你需求自己装置好 MySQL 数据库。

咱们先将项目所依靠的 Python 包导出，在项目根目录下履行：

pip freeze -l > requirements.txt

在项目根目录下新建 dockerfile 文件，写入以下内容：

FROM python:3.9-alpine
WORKDIR /flask_service
EXPOSE 5000
COPY . .
RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
ENV FLASK_DEBUG=True \
   FLASK_APP=run.py \
   FLASK_RUN_HOST=0.0.0.0 \
   FLASK_ENV=development \
   SECRET_KEY=b'#q)\\x00\xd6\x9f<iBQ\xd7;,\xe2E' \
   JWT_SECRET_KEY=b'#q)\\x00\xd6\x9f<iBQ\xd7;,\xe2E' \
   MYSQL_USER_NAME=<your mysql username> \
   MYSQL_USER_PASSWORD=<your mysql password> \
   MYSQL_HOSTNAME=<your mysql hostname> \
   MYSQL_PORT=<your mysql port> \
   MYSQL_DATABASE_NAME=<your database name>
RUN flask db migrate
RUN flask db upgrade
CMD ["flask", "run"]

留意，其间 <> 内的内容是需求填写成你自己的装备，由于咱们是运用 Docker 容器运转服务，所以你要保证你的容器内部能够衔接到数据库，不然或许会报错。 然后咱们运用指令行终端进入项目目录，履行打包指令：

sudo docker build -t flask_service_image .

打包完结后运转容器：

sudo docker run -d --restart=always -p 5000:5000 flask_service_image

至此，咱们整个开发到布置的流程就完结了，现在能够经过你服务器的 ip:5000/xxx 去调用接口了。

其他

假如你运用 Git 进行代码版本办理，那么我主张将以下文件疏忽：

*.pyc
.venv/
.env
migrations/versions/*

这里特别阐明一下 migrations/versions/* 这个疏忽途径。由于咱们运用的数据库搬迁插件是经过保护 alembic_version 表内的版本号来进行办理的，大多数情况下，咱们都是团队协同开发，并且本地数据库与线上数据库必定是分隔的，你和你的同事又是并行开发，所以肯定会导致版本号抵触，flask-migrate 扩展虽然也供给了处理抵触的 merge 指令，但咱们发现不太好用（或许是我运用办法不对，有研讨过的同学能够沟通一下）。所以最终咱们选用不提交版本号文件的办法，保障在上线时不会在数据库同步上花费太多时刻处理抵触。

总结

至此，咱们现已运用 Flask + Flask RESTful 快速建立了一个 API 服务，借助于 Flask 社区丰厚的扩展，咱们完结了 MySQL 数据库衔接、注册登录等基本接口服务，并且运用 Docker 将项目成功布置在服务器上。

尽管我在本文中尽或许地具体介绍了每一个过程和细节，可是难免会存在一些过错和不足之处。假如您在运用本文中介绍的办法时发现了任何过错或许有更好的办法，十分欢迎您纠正并提出主张，以便我能够不断改进和提高文章的质量。

我是荼锦，一个兴趣使然的开发者。十分感谢您阅览本文，期望本文对您有所协助！