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
情境
會員註冊系統.
- 新增會員
- 會員資料修改
- 刪除會員資料
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 <liang0816tw@gmail.com>
.. 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 <liang0816tw@gmail.com>
.. 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', 'paul@email.com', 'paul@password')
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': 'paul@email.com',
'password': 'paul@password',
'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://root@localhost/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等等就看實際狀況為何).