Background

之前寫過一篇類似的東西. 但隨著時間過去總覺得有更好的作法. 所以又心血來潮的在寫一篇文章記錄一下自己對 Flask Application 整體架構的想法. 那接著我會根據下面的想法來時做出一個簡單的Flask application.

Flask

• 利用Flask-script or Flask Cli 來run server & manage Flask service
• Route Manerger (List功能)
• Unittest for api (pytest)
• Artitechture for Flask service
• Doc
• Auth (csrf)
• No view in my Flask service
• Restful API Design
• DB module ORM
• 利用 gunicorn & gevent 來取代原生 WSGI
• 利用 Supervisor 來管理 Flask Process
• 利用 Logrotate 來優化 Log 管理

Database

• 獨立的專案, 專門管理DB migration.

CI

• 測試 & 部署
• 有名的第三方 service (簡單好用不用在自己架一個環境)
• 可串接github

情境

會員註冊系統.

1. 新增會員
2. 會員資料修改
3. 刪除會員資料

Setting

basic

$ mkdir app
$ cd app
$ virtualenv .venv
$ source .venv/bin/activate
$ mkdir src docs tests migration src/models src/auth
$ touch .gitignore run.py README.md config.py src/`__init__.py` tests/`__init__.py` src/models/`__init__.py` src/auth/`__init__.py` tests/conftest.py
$ pip3 install Flask Flask-CLI Flask-RESTful Flask-SQLAlchemy pytest PyJWT passlib SQLAlchemy alembic psycopg2-binary
$ pip3 freeze > requirements.txt

config.py

class Config(object): 
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    # change API_SECRET_KEY for your own appliaction
    API_SECRET_KEY = '4p2itrn3k9axbz4ts8f92vlzw4m30949k9sg5cum'
    # change LOGIN_SECRET_KEY for your own appliaction
    LOGIN_SECRET_KEY = 'qjugiycmyfgqhqfvmnpw8qntlewxexp7xghiwvi7'

class ProductionConfig(Config):
    DEBUG = False
    DOMAIN = 'http://127.0.0.1:5000'
    SQLALCHEMY_DATABASE_URI = 'sqlite:///'

class DevelopmentConfig(Config):
    DEVELOPMENT = True
    DEBUG = True
    DOMAIN = 'http://127.0.0.1:5000'
    SQLALCHEMY_DATABASE_URI = 'postgresql://taiker:@localhost/console'

class TestingConfig(Config):
    TESTING = True
    DEBUG = True
    DOMAIN = 'http://127.0.0.1:5000'
    SQLALCHEMY_DATABASE_URI = 'sqlite:///../flask.db'


config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,

    'default': DevelopmentConfig
}

run.py

"""
.. module:: run
   :platform: Ubuntu 16.04 & Mac OS
   :synopsis: Main function of flask service

.. moduleauthor:: Paul Liang <[email protected]>
.. date:: 2018-03-29
"""

from src import app, db
from flask import url_for
from config import config
import urllib
import click
import sys
import getopt
import os


@app.cli.command()
def initdb():
    """Initialize the database."""
    click.echo('Init the db')
    db.create_all()


@app.cli.command()
def renewdb():
    """Remove the database."""
    click.echo('Renew the db')
    db.drop_all()
    db.create_all()


@app.cli.command()
def list_routes():
    """List all routes on flask service"""
    output = []
    for rule in app.url_map.iter_rules():
        methods = ','.join(rule.methods)
        line = urllib.parse.unquote(
                "{:50s} {:20s} {}".format(rule.endpoint, methods, rule))
        output.append(line)

    for line in sorted(output):
        print(line)
        
if __name__ == '__main__':
    app.run()

mirgration

$ cd migration
$ alembic init alembic

設定檔的DB url 要與 config.py 裡的一致.

alembic.ini

...
sqlalchemy.url = postgresql://taiker:@localhost/console
...

src

src/__init__.py

from config import config
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config.from_object(config['default'])
db = SQLAlchemy(app)

基本設定完就來執行一下程式, 如果都沒問題就會看到以下畫面:

$ python3 run.py
 * Serving Flask app "src" (lazy loading)
 * Environment: production
   WARNING: Do not use the development server in a production environment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 196-623-139

Application

model

src/models/users.py

user.py 中 table column 的定義. 這會對應到 DB 實際上的欄位. 我們會透過 Flask-cli 來初始化DB. 之後如果DB欄位有修改的話我們才會透過 alembic來完成. 其餘的code則包含 檢查使用者密碼 & login & api token 驗證.

"""
.. module:: users
   :platform: Ubuntu 16.04, linux & Mac OS
   :synopsis: mapping to user table in database

.. moduleauthor:: Paul Liang <[email protected]>
.. date:: 2018-03-16
"""

from src import app
from src import db
from passlib.apps import custom_app_context as pwd_context
from sqlalchemy import Column, Integer, String, DateTime
from pprint import pprint
import jwt
import datetime


class User(db.Model):

    __tablename__ = 'users'

    # table columns
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True, index=True)
    password = Column(String)
    api_token = Column(String)
    status = Column(Integer)
    create_time = Column(DateTime)
    last_login = Column(DateTime)

    def __init__(self, name, email, password):
        self.name = name
        self.email = email
        self.password = self.hash_password(password)
        self.create_time = datetime.datetime.now()
        self.last_login = datetime.datetime.now()
        self.api_token = ''
        self.status = 0

    def hash_password(self, password):
        """
        hash user password
        :return: string
        """
        return pwd_context.hash(password + app.config['LOGIN_SECRET_KEY'])

    def verify_password(self, password):
        """
        get user password from db and hash it then ruturn.
        :return: string
        """
        return pwd_context.verify(
            password + app.config['LOGIN_SECRET_KEY'], self.password)

    def encode_login_auth_token(self):
        """
        Generates the auth token for login
        :return: string
        """
        try:
            payload = {
                'exp': datetime.datetime.now() +
                datetime.timedelta(hours=6, seconds=0),
                'iat': datetime.datetime.now(),
                'sub': self.email
            }

            return jwt.encode(
                payload,
                app.config['LOGIN_SECRET_KEY'],
                algorithm='HS256'
            )

        except Exception as e:
            return e

    @staticmethod
    def decode_login_auth_token(auth_token):
        """
        Decodes the auth token for login
        :param auth_token:
        :return: integer|string
        """
        try:
            payload = jwt.decode(auth_token, app.config['LOGIN_SECRET_KEY'], algorithms=['HS256'])
            user = User.query.filter_by(email=payload['sub']).first()
            return user
        except jwt.ExpiredSignatureError:
            return 'Signature expired. Please log in again.'
        except jwt.InvalidTokenError:
            return 'Invalid token. Please log in again.'

    def encode_api_auth_token(self):
        """
        Generates the auth token for api
        :return: string
        """
        try:
            payload = {
                'exp': datetime.datetime.now() +
                datetime.timedelta(hours=6, seconds=0),
                'iat': datetime.datetime.now(),
                'sub': self.email
            }

            return jwt.encode(
                payload,
                app.config['API_SECRET_KEY'],
                algorithm='HS256'
            )

        except Exception as e:
            return e

    @staticmethod
    def decode_api_auth_token(auth_token):
        """
        Decodes the auth token for api
        :param auth_token:
        :return: integer|string
        """
        try:
            payload = jwt.decode(auth_token, app.config['API_SECRET_KEY'], algorithms=['HS256'])
            user = User.query.filter_by(email=payload['sub']).first()
            return user
        except jwt.ExpiredSignatureError:
            return 'Signature expired. Please log in again.'
        except jwt.InvalidTokenError:
            return 'Invalid token. Please log in again.'

    def renew_last_login_date(self):
        """
        record user last login time
        :return: None
        """
        self.last_login = datetime.datetime.now()
        db.session.commit()

    def renew_api_token(self, api_token):
        """
        record user new api token
        :return: None
        """
        self.api_token = api_token
        db.session.commit()

auth

這部分的code則是

src/auth/auth.py

from functools import wraps
from flask import request, jsonify, abort, redirect, url_for
from src.models import users
from src import app


def check_auth(email, api_token):
    """
    Decodes the auth token for api request
    :param email:
    :param api_token:
    :return: integer
    """
    res = users.User.decode_api_auth_token(api_token)
    if isinstance(res, str):
        return 0
    else:
        return res.email == email


def requires_auth(f):
    """
    auth for api request
    :return: function object
    """
    @wraps(f)
    def decorated(*args, **kwargs):
        auth = request.authorization
        if not auth or not check_auth(auth.username, auth.password):
            abort(401)
        return f(*args, **kwargs)
    return decorated

login

run.py

# add this line to import section
from src.models import users

src/base.py

import sys
import os
from flask import jsonify, redirect, request, url_for, render_template
from src import app
from src import db
from src.auth import auth
from src.models import users
from pprint import pprint


@app.route("/", methods=['GET'])
def hello():
    return jsonify({'err': 0, 'err_msg': 'Welcome to Flask!'})

@app.route("/login", methods=['POST'])
def login():
    email = request.form['email']
    password = request.form['password']
    login_token = request.form['login_token']

    if login_token:
        res = users.User.decode_login_auth_token(login_token)
        if isinstance(res, str):
            return jsonify({'err': -1, 'err_msg': res})
        else:
            if res.email == email:
                api_token = (user.encode_api_auth_token()).decode("utf-8")
                res.renew_last_login_date()
                res.renew_api_token(api_token)
                return jsonify({'err': 0, 'err_msg': '',
                                'api_token': api_token})
            else:
                return jsonify({'err': -1, 'err_msg': 'Invalid token.'})

    user = users.User.query.filter_by(email=email).first()

    if user:
        pw_res = user.verify_password(password)
        if pw_res:
            login_token = (user.encode_login_auth_token()).decode("utf-8")
            api_token = (user.encode_api_auth_token()).decode("utf-8")
            user.renew_last_login_date()
            user.renew_api_token(api_token)
            res = jsonify({'err': 0, 'err_msg': '', 'login_token': login_token, 'api_token': api_token})
            return res
        else:
            return jsonify({'err': -1, 'err_msg': 'Invalid password.'})
    else:
        return jsonify({'err': -1, 'err_msg': 'Invalid email.'})

完成之後加入下面這行code到 src/__init__.py

src/__init__.py

...
# add this line to the end of the file.
from src import base

test

conftest.py 主要是初始化要測試前DB的初始值以方便測試.

tests/conftest.py

import pytest
from src import app, db
from src.models import users

@pytest.fixture(scope='session')
def user():
    print("user")
    user = users.User('paul', '[email protected]', '[email protected]')
    db.session.add(user)
    db.session.commit()
    return user

@pytest.fixture(scope='session')
def app_client():
    return app.test_client()

使用pytest時. 要測試的檔案 檔名的部分結尾or開頭要有test這個keyword

tests/test_base.py

import json
import pytest
import base64
import unittest
from src import app, db
from src.models import users
from pprint import pprint


def test_login(app_client, user):
    user = db.session.merge(user)
    data = {
        'email': '[email protected]',
        'password': '[email protected]',
        'login_token': ''
    }

    response = app_client.post('/login', data=data)
    print(json.loads(response.get_data(as_text=True)))
    res = json.loads(response.get_data(as_text=True))
    assert res['err'] == 0

cli

ok 最後回到 root 資料夾. 我們要設定 flask-cli 指令來幫助我們測試. 指令的code都在run.py中有興趣的人可以看看.

$ export FLASK_APP=./run.py
# init db完之後我們可以看db是否有新增出一個users table.
$ flask initdb
$ $ flask list-routes
hello                                       GET,OPTIONS,HEAD     /
login                                       OPTIONS,POST         /login
static                                      lsGET,OPTIONS,HEAD     /static/<path:filename>
$ flask test
Renew the db
========================= test session starts ========================
platform darwin -- Python 3.7.0, pytest-3.9.3, py-1.7.0, pluggy-0.8.0
rootdir: /Users/taiker/Google_Drive/Dev/xinyisheng-erp-flask, inifile:
collected 5 items

tests/test_base.py .                                            [ 20%]
tests/models/test_users.py ....                                 [100%]

===================== 5 passed in 6.41 seconds =======================

gunicorn & gevent

Gunicorn 是一個 Python WSGI UNIX 的 HTTP 服務器. 我們透過來讓我們整體的Flask service效能變得更好.

gevent 是一個 coroutine 的優化程式. 比起 Gunicorn 預設的 sync 模式有更好的效能.

install

$ sudo pip3 install gunicorn
$ sudo pip3 install gevent

gunicorn 的配置文件

gunicorn.conf

import multiprocessing

bind = "0.0.0.0:8000"

workers = multiprocessing.cpu_count() * 2 + 1
worker_class = 'gevent'

max_requests = 1000
timeout = 30
keep_alive = 2

preload = True
loglevel = 'info'
pidfile = '/tmp/gunicorn.pid'
errorlog = '/var/log/app/error.log'
accesslog = '/var/log/app/access.log'
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s"'

這樣基本上就完成了, 接下來我們會透過Supervisor來啟動gunicorn

Supervisor

install

$ sudo apt-get install supervisor

conf

/etc/supervisor/conf.d

[program:app]
directory=/var/www/app
command=/var/www/app/.venv/bin/gunicorn run:app -c /var/www/app/gunicorn.conf

autostart=true
autorestart=true
stderr_logfile=/var/log/supervisor/app.err.log
stdout_logfile=/var/log/supervisor/app.out.log

stdout_logfile_maxbytes=10MB
stdout_logfile_backups=10
stdout_capture_maxbytes=10MB
stdout_events_enabled=false

stderr_logfile_maxbytes=10MB
stderr_logfile_backups=10
stderr_capture_maxbytes=10MB

接著我們就可以藉由 supervisor 的指令來幫助我們管理整個 Flask Service.

log rotate

接著我們透過 logrotate 來優化我們所產出的log

$ vim /etc/logrotate.d/app.conf

/etc/logrotate.d/app.conf

/var/log/app/access.log {
    rotate 10
    daily
    copytruncate
    compress
    missingok
    create 644 root root
    dateext
}

/var/log/app/error.log {
    rotate 10
    daily
    copytruncate
    compress
    missingok
    create 644 root root
    dateext
}

檔案編輯完後接著我們執行下面的指令. 再到Log檔的目錄下看優該就可以看出有log rotate的變化.

$ sudo logrotate -f -v /etc/logrotate.d/aoo

Document

api docment 我們利用 apiDoc 這個 Tool 來幫助我們. 詳細內容可以參考我之前的blog文章

DB migration

那接著如果我們想要變動DB的話 我們需要透過以下步驟來完成:

$ cd migration
$ alembic revision -m "modify users table"
Generating /path/to/migration/alembic/versions/8b8b6c48137d_modify_users_table.py ... done
$ vim alembic/versions/8b8b6c48137d_modify_users_table.py

alembic/versions/8b8b6c48137d_modify_users_table.py

"""modify users table

Revision ID: 8b8b6c48137d
Revises:
Create Date: 2018-10-31 16:53:07.771186

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '8b8b6c48137d'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    op.add_column('users',
        Column('new_column', String())
    )

def downgrade():
    op.drop_column('users','new_column')

那要記得 upgrade() 是完成你想要修正DB的動作. downgrade() 則是妳這次動作的還原過程. 這樣才能確保之後如果DB變動有問題的話才能復原回上一個版本.

接著我們執行 upgrad 指令, head 則表示是最新的一個 revision.

$ alembic upgrade head

完成後檢查一下DB. 確定有多了一個欄位則表示成功.

console=# \d users
                                         Table "public.users"
   Column    |            Type             | Collation | Nullable |              Default
-------------+-----------------------------+-----------+----------+-----------------------------------
 id          | integer                     |           | not null | nextval('users_id_seq'::regclass)
 name        | character varying           |           |          |
 email       | character varying           |           |          |
 password    | character varying           |           |          |
 api_token   | character varying           |           |          |
 status      | integer                     |           |          |
 create_time | timestamp without time zone |           |          |
 last_login  | timestamp without time zone |           |          |
 new_column  | character varying           |           |          |
Indexes:
    "users_pkey" PRIMARY KEY, btree (id)
    "ix_users_email" UNIQUE, btree (email)

保持一個好習慣. 測試完 upgrad() 也要順便測試一下 downgrad()

$ alembic downgrade -1

都沒問題的話下一個section我們要進入CI的部分.

CI (circleCI)

$ mkdir .circleci
$ cd .circleci
$ vim config.yml

.circleci/config.yml

version: 2
jobs:
  build:
    working_directory: ~/app
    docker:
      - image: circleci/python:3.6.4
        environment:
          PIPENV_VENV_IN_PROJECT: true
          DATABASE_URL: postgresql://[email protected]/circle_test?sslmode=disable
      - image: circleci/postgres:9.6.2
        environment:
          POSTGRES_USER: taiker
          POSTGRES_DB: console

    steps:
      - checkout
      - restore_cache:
          key: deps1-{{ .Branch }}-{{ checksum "requirements.txt" }}
      - run: sudo apt-get install python3-pip
      - run: sudo pip3 install --upgrade pip
      - run:
          command: |
            python3 -m venv venv
            . venv/bin/activate
            sudo pip3 install -r requirements.txt
      - run:
          command: |
            export FLASK_APP=/home/circleci/app/run.py
            flask renewdb
            pytest

裡面的內容可依個人實際狀況&喜好自己編輯. (POSTGRES_USER, POSTGRES_DB等等就看實際狀況為何).

Reference

• https://www.youtube.com/watch?v=1X3_gQwfabI
• https://flask-migrate.readthedocs.io/en/latest/
• https://flask-migrate.readthedocs.io/en/latest/
• https://blog.burn-i.com/20180418/flask-migrate/