2024年5月4日 星期六

Flask網頁開發 - 第二版 - 應用程式開發介面、測試與部署 (6)

　　來說明這本由Miguel Grinberg著作，賴屹民翻譯的《Flask網頁開發 第二版》(Flask Web Development, 2nd Edition)，本書的撰寫方式如同網頁開發人員新增應用程式功能，每新增一項功能，就會有新的章節說明這樣的新功能，是使用什麼樣的技術製作而成；由於網頁應用程式具結構性，程式碼之間的撰寫與修改是息息相關的，故作者有提供Repository供讀者檢視與試驗。

　　我們可以依照本書各章節的順序，逐一檢視各章節在Repository對應的 Branch，藉由各個Branch的切換與推移，可以學習、理解與推演作者修改程式的邏輯，這對於Flask的學習是非常重要的。

　　「Flask網頁開發」系列網誌，整理的方向也只是羅列修改方向一致的各個Branch，統整為較容易查找的功能主題，並輔以註解說明特殊模組、套件的API如何使用，以及使用的目的，對於完全沒有網頁應用程式開發經驗的讀者來說，依舊是極度難以理解的一系列筆記，建議還是得先學習好Flask的基礎，再掌握好本書作者訂定的目錄框架，最後才能掌握住本書的程式開發方式。

第二部分、範例：社群部落格app

第十四章、應用程式開發介面－API(14a)

l   API：
git checkout 14a

l   本書的API藍圖結構：
|-flasky
  |-app/
    |-api
      |-__init__.py
      |-users.py
      |-posts.py
      |-comments.py
      |-authentication.py
      |-errors.py
      |-decorators.py

l   安裝Flask-HTTPAuth以驗證使用者，使用HTTP身分驗證時，使用者憑證會放在所有request的Authorization標頭中：
pip install flask-httpauth

l   用HTTPie測試web服務，在命令列測試Python web服務最常用的兩種用戶端是cURL與HTTPie，而HTTPie是專為API請求量身訂做的：
pip install httpie

l   app/__init__.py：
# ...
def create_app(config_name):
    # ...
    # 註冊API藍圖
    from .api import api as api_blueprint
    app.register_blueprint(api_blueprint, url_prefix="/api/v1")
    # ...

l   app/models.py：
# ...
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from flask import current_app, request, url_for
from flask_login import UserMixin, AnonymousUserMixin
from app.exceptions import ValidationError
from . import db, login_manager
# ...
class User(UserMixin, db.Model):
    # ...
    # 將使用者轉換成JSON可序列化字典
    def to_json(self):
        json_user = {
            "url": url_for("api.get_user", id=self.id),
            "username": self.username,
            "member_since": self.member_since,
            "last_seen": self.last_seen,
            "posts_url": url_for("api.get_user_posts", id=self.id),
            "followed_posts_url": url_for("api.get_user_followed_posts", id=self.id),
            "post_count": self.posts.count()
        }
        return json_user
   
    # 支援權杖驗證(Token-based Authentication)功能
    def generate_auth_token(self, expiration):
        s = Serializer(current_app.config["SECRET_KEY"], expires_in=expiration)
        return s.dumps({"id": self.id}).decode("utf-8")
   
    @staticmethod
    # 支援權杖驗證(Token-based Authentication)功能
    def verify_auth_token(token):
        s = Serializer(current_app.config["SECRET_KEY"])
        try:
            data = s.loads(token)
        except:
            return None
        return User.query.get(data["id"])
# ...
class Post(db.Model):
    # ...
    # 將文章轉換成JSON可序列化字典
    def to_json(self):
        json_post = {
            "url": url_for("api.get_post", id=self.id),
            "body": self.body,
            "body_html": self.body_html,
            "timestamp": self.timestamp,
            "author_url": url_for("api.get_user", id=self.author_id),
            "comments_url": url_for("api.get_post_comments", id=self.id),
            "comment_count": self.comments.count()
        }
        return json_post
   
    # 用JSON建立部落格文章
    @staticmethod
    def from_json(json_post):
        body = json_post.get("body")
        if body is None or body == "":
            raise ValidationError("post does not have a body")
        return Post(body=body)
# ...
class Comment(db.Model):
    # ...
    def to_json(self):
        json_comment = {
            "url": url_for("api.get_comment", id=self.id),
            "post_url": url_for("api.get_post", id=self.post_id),
            "body": self.body,
            "body_html": self.body_html,
            "timestamp": self.timestamp,
            "author_url": url_for("api.get_user", id=self.author_id),
        }
        return json_comment
   
    @staticmethod
    def from_json(json_comment):
        body = json_comment.get("body")
        if body is None or body == "":
            raise ValidationError("comment does not have a body")
        return Comment(body=body)

l   app/exceptions.py：
# 將ValidationError類別簡單地寫成Python ValueError的子類別
class ValidationError(ValueError):
    pass

l   app/main/errors.py：
# 處理404與500 HTTP狀態碼比較麻煩，因為這些錯誤通常是Flask自己產生的，而且它會回傳一個HTML回應，而不是JSON，這可能會造成用戶端的混亂
from flask import render_template, request, jsonify
from . import main

# 使用HTTP內容協商(Content Negotiation)的403錯誤處理常式
@main.app_errorhandler(403)
def forbidden(e):
    if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
        response = jsonify({"error": "forbidden"})
        response.status_code = 403
        return response
    return render_template("403.html"), 403

# 使用HTTP內容協商(Content Negotiation)的404錯誤處理常式
@main.app_errorhandler(404)
def page_not_found(e):
    if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
        response = jsonify({"error": "not found"})
        response.status_code = 404
        return response
    return render_template("404.html"), 404

# 使用HTTP內容協商(Content Negotiation)的500錯誤處理常式
@main.app_errorhandler(500)
def internal_server_error(e):
    if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
        response = jsonify({"error": "internal server error"})
        response.status_code = 500
        return response
    return render_template("500.html"), 500

l   app/api/__init__.py：
# 建立API藍圖
from flask import Blueprint
api = Blueprint("api", __name__)
from . import authentication, posts, users, comments, errors

l   app/api/errors.py：
from flask import jsonify
# 本書作者自訂的ValidationError類別，位於app/exceptions.py
from app.exceptions import ValidationError
from . import api

# 狀態碼400的API錯誤處理常式
def bad_request(message):
    response = jsonify({"error": "forbidden", "message": message})
    response.status_code = 400
    return response

# 狀態碼401的API錯誤處理常式
def unauthorized(message):
    response = jsonify({"error": "unauthorized", "message": message})
    response.status_code = 401
    return response

# 狀態碼403的API錯誤處理常式
def forbidden(message):
    response = jsonify({"error": "forbidden", "message": message})
    response.status_code = 403
    return response

# app/exceptions.py的ValidationError異常的API錯誤處理常式
# 被裝飾的函式會在指定類別發出異常時執行
@api.errorhandler(ValidationError)
def validation_error(e):
    # e.args[0]為取得Exception詳細內容
    return bad_request(e.args[0])

l   app/api/authentication.py：
from flask import g, jsonify
from flask_httpauth import HTTPBasicAuth
from ..models import User
from . import api
from .errors import unauthorized, forbidden

# 初始化Flask-HTTPAuth
auth = HTTPBasicAuth()

@auth.verify_password
def verify_password(email_or_token, password):
    if email_or_token == "":
        return False
    # 如果密碼是空的，我們假設email_or_token欄位是權杖並進行驗證
    if password == "":
        g.current_user = User.verify_auth_token(email_or_token)
        g.token_used = True
        return g.current_user is not None
    user = User.query.filter_by(email=email_or_token.lower()).first()
    if not user:
        return False
    g.current_user = user
    g.token_used = False
    return user.verify_password(password)

@auth.error_handler
def auth_error():
    return unauthorized("Invalid credentials")

@api.before_request
@auth.login_required
def before_request():
    if not g.current_user.is_anonymous and not g.current_user.confirmed:
        return forbidden("Unconfirmed account")

# 產生身分驗證權杖
@api.route("/tokens/", methods=["POST"])
def get_token():
    if g.current_user.is_anonymous or g.token_used:
        return unauthorized("Invalid credentials")
    return jsonify({"token": g.current_user.generate_auth_token(expiration=3600), "expiration": 3600})

l   app/api/posts.py：
from flask import jsonify, request, g, url_for, current_app
from .. import db
from ..models import Post, Permission
from . import api
from .decorators import permission_required
from .errors import forbidden

# 文章的GET資源處理常式
@api.route("/posts/")
def get_posts():
    page = request.args.get("page", 1, type=int)
    pagination = Post.query.paginate(page=page, per_page=current_app.config["FLASKY_POSTS_PER_PAGE"], error_out=False)
    posts = pagination.items
    prev = None
    if pagination.has_prev:
        prev = url_for("api.get_posts", page=page-1)
    next = None
    if pagination.has_next:
        next = url_for("api.get_posts", page=page+1)
    return jsonify({
        "posts": [post.to_json() for post in posts],
        "prev": prev,
        "next": next,
        "count": pagination.total
    })
# def get_posts():
#     posts = Post.query.all()
#     return jsonify({"posts": [post.to_json() for post in posts]})

# 文章的GET資源處理常式
@api.route("/posts/<int:id>")
def get_post(id):
    post = Post.query.get_or_404(id)
    return jsonify(post.to_json())

# 文章的POST資源處理常式
@api.route("/posts/", methods=["POST"])
@permission_required(Permission.WRITE)
def new_post():
    post = Post.from_json(request.json)
    post.author = g.current_user
    db.session.add(post)
    db.session.commit()
    return jsonify(post.to_json()), 201, {"Location": url_for("api.get_post", id=post.id)}

# 文章的PUT資源處理常式
@api.route("/posts/<int:id>", methods=["PUT"])
@permission_required(Permission.WRITE)
def edit_post(id):
    post = Post.query.get_or_404(id)
    if g.current_user != post.author and not g.current_user.can(Permission.ADMIN):
        return forbidden("Insufficient permissions")
    post.body = request.json.get("body", post.body)
    db.session.add(post)
    db.session.commit()
    return jsonify(post.to_json())

l   app/api/users.py：
from flask import jsonify, request, current_app, url_for
from . import api
from ..models import User, Post

@api.route("/users/<int:id>")
def get_user(id):
    user = User.query.get_or_404(id)
    return jsonify(user.to_json())

@api.route("/users/<int:id>/posts/")
def get_user_posts(id):
    user = User.query.get_or_404(id)
    page = request.args.get("page", 1, type=int)
    pagination = user.posts.order_by(Post.timestamp.desc()).paginate(page=page, per_page=current_app.config["FLASKY_POSTS_PER_PAGE"], error_out=False)
    posts = pagination.items
    prev = None
    if pagination.has_prev:
        prev = url_for("api.get_user_posts", id=id, page=page-1)
    next = None
    if pagination.has_next:
        next = url_for("api.get_user_posts", id=id, page=page+1)
    return jsonify({
        "posts": [post.to_json() for post in posts],
        "prev": prev,
        "next": next,
        "count": pagination.total
    })

@api.route("/users/<int:id>/timeline/")
def get_user_followed_posts(id):
    user = User.query.get_or_404(id)
    page = request.args.get("page", 1, type=int)
    pagination = user.followed_posts.order_by(Post.timestamp.desc()).paginate(page=page, per_page=current_app.config["FLASKY_POSTS_PER_PAGE"], error_out=False)
    posts = pagination.items
    prev = None
    if pagination.has_prev:
        prev = url_for("api.get_user_followed_posts", id=id, page=page-1)
    next = None
    if pagination.has_next:
        next = url_for("api.get_user_followed_posts", id=id, page=page+1)
    return jsonify({
        "posts": [post.to_json() for post in posts],
        "prev": prev,
        "next": next,
        "count": pagination.total
    })

l   app/api/comments.py：
from flask import jsonify, request, g, url_for, current_app
from .. import db
from ..models import Post, Permission, Comment
from . import api
from .decorators import permission_required

@api.route("/comments/")
def get_comments():
    page = request.args.get("page", 1, type=int)
    pagination = Comment.query.order_by(Comment.timestamp.desc()).paginate(page=page, per_page=current_app.config["FLASKY_COMMENTS_PER_PAGE"], error_out=False)
    comments = pagination.items
    prev = None
    if pagination.has_prev:
        prev = url_for("api.get_comments", page=page-1)
    next = None
    if pagination.has_next:
        next = url_for("api.get_comments", page=page+1)
    return jsonify({
        "comments": [comment.to_json() for comment in comments],
        "prev": prev,
        "next": next,
        "count": pagination.total
    })

@api.route("/comments/<int:id>")
def get_comment(id):
    comment = Comment.query.get_or_404(id)
    return jsonify(comment.to_json())

@api.route("/posts/<int:id>/comments/")
def get_post_comments(id):
    post = Post.query.get_or_404(id)
    page = request.args.get("page", 1, type=int)
    pagination = post.comments.order_by(Comment.timestamp.asc()).paginate(page=page, per_page=current_app.config["FLASKY_COMMENTS_PER_PAGE"], error_out=False)
    comments = pagination.items
    prev = None
    if pagination.has_prev:
        prev = url_for("api.get_post_comments", id=id, page=page-1)
    next = None
    if pagination.has_next:
        next = url_for("api.get_post_comments", id=id, page=page+1)
    return jsonify({
        "comments": [comment.to_json() for comment in comments],
        "prev": prev,
        "next": next,
        "count": pagination.total
    })

@api.route("/posts/<int:id>/comments/", methods=["POST"])
@permission_required(Permission.COMMENT)
def new_post_comment(id):
    post = Post.query.get_or_404(id)
    comment = Comment.from_json(request.json)
    comment.author = g.current_user
    comment.post = post
    db.session.add(comment)
    db.session.commit()
    return jsonify(comment.to_json()), 201, \
        {"Location": url_for("api.get_comment", id=comment.id)}

l   app/api/decorators.py：
from functools import wraps
from flask import g
from .errors import forbidden

# permission_required裝飾器
def permission_required(permission):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not g.current_user.can(permission):
                return forbidden('Insufficient permissions')
            return f(*args, **kwargs)
        return decorated_function
    return decorator

l   tests/test_user_model.py：
import unittest
from app import create_app, db
# ...
class UserModelTestCase(unittest.TestCase):
    # ...
    def test_to_json(self):
        u = User(email="john@example.com", password="cat")
        db.session.add(u)
        db.session.commit()
        with self.app.test_request_context("/"):
            json_user = u.to_json()
        expected_keys = ["url", "username", "member_since", "last_seen", "posts_url", "followed_posts_url", "post_count"]
        self.assertEqual(sorted(json_user.keys()), sorted(expected_keys))
        self.assertEqual("/api/v1/users/" + str(u.id), json_user["url"])

l   本書API使用範例，用HTTPie測試web服務的GET與POST：
http --json --auth <email>:<password> GET http://localhost:5000/api/v1/posts/

http --json --auth <email>:<password> POST http://localhost:5000/api/v1/posts/ "body=I'm adding a post from the *command line*."

l   本書API使用範例，用HTTPie測試web服務，若要使用身分驗證權杖，先送一個POST request給/api/v1/tokens/，取得權杖後，接著對使用者名稱欄位傳入回傳的權杖，留空密碼欄位：
http --json --auth <email>:<password> POST http://localhost:5000/api/v1/tokens/

http --json --auth <token>: GET http://localhost:5000/api/v1/posts/

第三部分、最後一哩路

第十五章、測試－覆蓋率指標(15a)

l   覆蓋率指標：
git checkout 15a

l   測試程式覆蓋率工具，這個工具附帶一個命令列腳本可在啟用程式碼覆蓋率工具的情況下執行任何Python App，也可以用程式來啟動覆蓋率引擎：
pip install coverage

l   flasky.py：
import os

COV = None
if os.environ.get("FLASK_COVERAGE"):
    import coverage
    # 啟動覆蓋率引擎，branch=True選項會啟用分支覆蓋分析，追蹤有哪幾行程式碼被執行，也會確認每一個條件式是否已經執行了True與False兩種情況
    # 如果沒有include選項，在虛擬環境安裝的所有擴充套件以及測試程式本身都會被納入覆蓋率報告，讓報告有許多雜訊
    COV = coverage.coverage(branch=True, include="app/*")
    COV.start()

import sys
# Click模組是製作命令列的工具，可參考https://myapollo.com.tw/blog/python-click/的介紹
import click
from flask_migrate import Migrate
from app import create_app, db
from app.models import User, Follow, Role, Permission, Post, Comment

app = create_app(os.getenv("FLASK_CONFIG") or "default")
migrate = Migrate(app, db)

@app.shell_context_processor
def make_shell_context():
    return dict(db=db, User=User, Follow=Follow, Role=Role, Permission=Permission, Post=Post, Comment=Comment)

@app.cli.command()
# 在flask test命令傳入--coverage選項來啟用程式碼覆蓋率的支援
# 下flask test --coverage指令，可得純文字報告的範例
# 讓click用引數將布林旗標值傳給函式
@click.option("--coverage/--no-coverage", default=False, help="Run tests under code coverage.")
@click.argument("test_names", nargs=-1)
def test(coverage, test_names):
    """Run the unit tests."""
    if coverage and not os.environ.get("FLASK_COVERAGE"):
        import subprocess
        os.environ["FLASK_COVERAGE"] = "1"
        sys.exit(subprocess.call(sys.argv))
        # os.exec*家族主要用來代替目前程序，執行新的而不返回值
        # sys.executable取得可執行Python的exe的絕對路徑
        # os.execvp(sys.executable, [sys.executable] + sys.argv)
    import unittest
    if test_names:
        tests = unittest.TestLoader().loadTestsFromNames(test_names)
    else:
        # TestLoader()加載測試實例，並將它們返回給測試套件
        # discover()找到指定目錄下的所有測試腳本
        tests = unittest.TestLoader().discover("tests")
    # verbosity=2為詳細模式，測試結果會顯示每個測試實例所有訊息
    unittest.TextTestRunner(verbosity=2).run(tests)
    if COV:
        COV.stop()
        COV.save()
        print("Coverage Summary:")
        COV.report()
        basedir = os.path.abspath(os.path.dirname(__file__))
        covdir = os.path.join(basedir, "tmp/coverage")
        COV.html_report(directory=covdir)
        print("HTML version: file://%s/index.html" % covdir)
        COV.erase()

第十五章、測試－使用Flask測試用戶端進行單元測試(15b)

l   使用Flask測試用戶端進行單元測試：
git checkout 15b

l   測試用戶端(Test Client)會複製Web伺服器執行app時的環境，讓測試程式可以扮演用戶端的角色，並傳送Request。

l   config.py：
# Flask-WTF生成的所有表單都有一個隱藏欄位，裡面儲存了必須連同表單一起送出的CSRF權杖，必須解析回應的HTML來取出權杖
# ...
# 在測試組態中停用CSFR保護
class TestingConfig(Config):
    # ...
    WTF_CSRF_ENABLED = False
# ...

l   tests/test_client.py：
import re
import unittest
from app import create_app, db
from app.models import User, Role

# 使用Flask測試用戶端的測試框架
class FlaskClientTestCase(unittest.TestCase):
    def setUp(self):
        self.app = create_app("testing")
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()
        # self.client是Flask測試用戶端物件
        self.client = self.app.test_client(use_cookies=True)
   
    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self.app_context.pop()
   
    def test_home_page(self):
        response = self.client.get("/")
        self.assertEqual(response.status_code, 200)
        # get_data()預設以位元組陣列來回傳，as_text=True轉換成字串
        # self.assertTrue(b"Stranger" in response.data)
        self.assertTrue("Stranger" in response.get_data(as_text=True))
   
    # 用Flask測試用戶端來模擬新使用者的操作流程
    def test_register_and_login(self):
        # 註冊新帳號
        response = self.client.post("/auth/register", data={
            "email": "john@example.com",
            "username": "john",
            "password": "cat",
            "password2": "cat"
        })
        self.assertEqual(response.status_code, 302)
       
        # 用新帳號登入
        # follow_redirects=True引數來讓測試用戶端扮演瀏覽器，並自動發出一個轉址URL GET request
        response = self.client.post("/auth/login", data={
            "email": "john@example.com",
            "password": "cat"
        }, follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        # Jinja2模板的建立方式，在最終HTML中，會有額外的空格，故使用正規表達式
        self.assertTrue(re.search("Hello,\s+john!", response.get_data(as_text=True)))
        self.assertTrue("You have not confirmed your account yet" in response.get_data(as_text=True))
       
        # 傳送確認權杖
        user = User.query.filter_by(email="john@example.com").first()
        token = user.generate_confirmation_token()
        response = self.client.get("/auth/confirm/{}".format(token), follow_redirects=True)
        user.confirm(token)
        self.assertEqual(response.status_code, 200)
        self.assertTrue("You have confirmed your account" in response.get_data(as_text=True))
       
        # 登出
        response = self.client.get("/auth/logout", follow_redirects=True)
        self.assertEqual(response.status_code, 200)
        self.assertTrue("You have been logged out" in response.get_data(as_text=True))

第十五章、測試－使用Flask測試用戶端進行API測試(15c)

l   使用Flask測試用戶端進行API測試：
git checkout 15c

l   tests/test_api.py：
import unittest
import json
import re
from base64 import b64encode
from app import create_app, db
from app.models import User, Role, Post, Comment

# 使用Flask測試用戶端來做RESTful API測試
class APITestCase(unittest.TestCase):
    def setUp(self):
        self.app = create_app("testing")
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()
        self.client = self.app.test_client()
   
    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self.app_context.pop()
   
    # 會回傳必須與多數API request一起傳送的標頭，包括身分驗證憑證與MIME型態相關的標頭
    def get_api_headers(self, username, password):
        return {
            "Authorization": "Basic " + b64encode((username + ":" + password).encode("utf-8")).decode("utf-8"),
            "Accept": "application/json",
            "Content-Type": "application/json"
        }
   
    def test_404(self):
        response = self.client.get("/wrong/url", headers=self.get_api_headers("email", "password"))
        self.assertEqual(response.status_code, 404)
        json_response = json.loads(response.get_data(as_text=True))
        self.assertEqual(json_response["error"], "not found")
   
    # 確保不包含身分驗證憑證的request會被401錯誤碼拒絕
    def test_no_auth(self):
        response = self.client.get(url_for("api.get_posts"), content_type="application/json")
        # response = self.client.get("/api/v1/posts/", content_type="application/json")
        self.assertEqual(response.status_code, 401)
   
    def test_bad_auth(self):
        # 新增一名使用者
        r = Role.query.filter_by(name="User").first()
        self.assertIsNotNone(r)
        u = User(email="john@example.com", password="cat", confirmed=True, role=r)
        db.session.add(u)
        db.session.commit()
       
        # 以錯誤的密碼驗證
        response = self.client.get(
            "/api/v1/posts/",
            headers=self.get_api_headers("john@example.com", "dog"))
        self.assertEqual(response.status_code, 401)
   
    def test_token_auth(self):
        # 新增一名使用者
        r = Role.query.filter_by(name="User").first()
        self.assertIsNotNone(r)
        u = User(email="john@example.com", password="cat", confirmed=True, role=r)
        db.session.add(u)
        db.session.commit()
       
        # 以錯誤的Token發送Request
        response = self.client.get(
            "/api/v1/posts/",
            headers=self.get_api_headers("bad-token", ""))
        self.assertEqual(response.status_code, 401)
       
        # 取得Token
        response = self.client.post(
            "/api/v1/tokens/",
            headers=self.get_api_headers("john@example.com", "cat"))
        self.assertEqual(response.status_code, 200)
        json_response = json.loads(response.get_data(as_text=True))
        self.assertIsNotNone(json_response.get("token"))
        token = json_response["token"]
       
        # 以Token發送Request
        response = self.client.get(
            "/api/v1/posts/",
            headers=self.get_api_headers(token, ""))
        self.assertEqual(response.status_code, 200)
   
    def test_anonymous(self):
        response = self.client.get(
            "/api/v1/posts/",
            headers=self.get_api_headers("", ""))
        self.assertEqual(response.status_code, 401)
   
    def test_unconfirmed_account(self):
        # 新增一名未受驗證的使用者
        r = Role.query.filter_by(name="User").first()
        self.assertIsNotNone(r)
        u = User(email="john@example.com", password="cat", confirmed=False, role=r)
        db.session.add(u)
        db.session.commit()
        # 以未驗證帳戶取得貼文列表
        response = self.client.get(
            "/api/v1/posts/",
            headers=self.get_api_headers("john@example.com", "cat"))
        self.assertEqual(response.status_code, 403)
   
    # 將一位使用者加入資料庫，接著插入一篇部落格文章，接著讀回
    def test_posts(self):
        # 加入使用者
        r = Role.query.filter_by(name="User").first()
        self.assertIsNotNone(r)
        u = User(email="john@example.com", password="cat", confirmed=True, role=r)
        db.session.add(u)
        db.session.commit()
       
        # 撰寫一則空的文章
        response = self.client.post(
            "/api/v1/posts/",
            headers=self.get_api_headers("john@example.com", "cat"),
            data=json.dumps({"body": ""}))
        self.assertEqual(response.status_code, 400)
       
        # 撰寫文章
        response = self.client.post(
            "/api/v1/posts/",
            headers=self.get_api_headers("john@example.com", "cat"),
            data=json.dumps({"body": "body of the *blog* post"}))
        self.assertEqual(response.status_code, 201)
        url = response.headers.get("Location")
        self.assertIsNotNone(url)
       
        # 取得新文章
        response = self.client.get(
            url,
            headers=self.get_api_headers("john@example.com", "cat"))
        self.assertEqual(response.status_code, 200)
        json_response = json.loads(response.get_data(as_text=True))
        self.assertEqual("http://localhost" + json_response["url"], url)
        self.assertEqual(json_response["body"], "body of the *blog* post")
        self.assertEqual(json_response["body_html"], "<p>body of the <em>blog</em> post</p>")
        json_post = json_response
       
        # 取得用戶的貼文
        response = self.client.get(
            "/api/v1/users/{}/posts/".format(u.id),
            headers=self.get_api_headers("john@example.com", "cat"))
        self.assertEqual(response.status_code, 200)
        json_response = json.loads(response.get_data(as_text=True))
        self.assertIsNotNone(json_response.get("posts"))
        self.assertEqual(json_response.get("count", 0), 1)
        self.assertEqual(json_response["posts"][0], json_post)
       
        # 以追蹤者的角色從用戶那裡取得貼文
        response = self.client.get(
            "/api/v1/users/{}/timeline/".format(u.id),
            headers=self.get_api_headers("john@example.com", "cat"))
        self.assertEqual(response.status_code, 200)
        json_response = json.loads(response.get_data(as_text=True))
        self.assertIsNotNone(json_response.get("posts"))
        self.assertEqual(json_response.get("count", 0), 1)
        self.assertEqual(json_response["posts"][0], json_post)
       
        # 編輯貼文
        response = self.client.put(
            url,
            headers=self.get_api_headers("john@example.com", "cat"),
            data=json.dumps({"body": "updated body"}))
        self.assertEqual(response.status_code, 200)
        json_response = json.loads(response.get_data(as_text=True))
        self.assertEqual("http://localhost" + json_response["url"], url)
        self.assertEqual(json_response["body"], "updated body")
        self.assertEqual(json_response["body_html"], "<p>updated body</p>")
        # ...其餘省略，詳見作者Repository

第十五章、測試－使用Selenium進行單元測試(15d)

l   使用Selenium進行單元測試：
git checkout 15d

l   用Selenium來做端對端測試：
pip install selenium

l   可以從ChromeDriver網站(https://sites.google.com/a/chromium.org/chromedriver/downloads)下載一般的ChromeDriver安裝程式。

l   app/main/views.py：
# ...
# 當所有測試都完成時，所有的Flask伺服器都必須停止；Werkzeug Web伺服器有關機選項，要求伺服器關機唯一的方法是傳送一般的HTTP Request
# Werkzeug Web伺服器關機路由
@main.route("/shutdown")
def server_shutdown():
    # 當app在測試模式下運行時，關機路由才有效
    if not current_app.testing:
        abort(404)
    # 實際的關機程序需要呼叫Werkzeug在環境中公開的關機函式，呼叫函式並從request回傳後，開發web伺服器就必須退場了
    shutdown = request.environ.get("werkzeug.server.shutdown")
    if not shutdown:
        abort(500)
    shutdown()
    return "Shutting down..."
# ...

l   tests/test_selenium.py：
# 用Selenium執行測試的框架
import re
import threading
import time
import unittest
from selenium import webdriver
from app import create_app, db, fake
from app.models import Role, User, Post

class SeleniumTestCase(unittest.TestCase):
    client = None
   
    # setUpClass和tearDownClass在整個class只會執行一次，setUp和tearDown在每個測試方法之前和之後都會運行
    @classmethod
    def setUpClass(cls):
        # 啟動Chrome
        options = webdriver.ChromeOptions()
        options.add_argument("headless")
        try:
            cls.client = webdriver.Chrome(chrome_options=options)
        except:
            pass
       
        # 如果瀏覽器無法啟動，就跳過這些測試
        if cls.client:
            # 建立app
            cls.app = create_app("testing")
            cls.app_context = cls.app.app_context()
            cls.app_context.push()
           
            # 禁止記錄，來讓unittest有簡明的輸出
            import logging
            logger = logging.getLogger("werkzeug")
            logger.setLevel("ERROR")
           
            # 建立資料庫並填入一些偽造資料
            db.create_all()
            Role.insert_roles()
            fake.users(10)
            fake.posts(10)
           
            # 加入一位管理員使用者
            admin_role = Role.query.filter_by(name="Administrator").first()
            admin = User(email="john@example.com", username="john", password="cat", role=admin_role, confirmed=True)
            db.session.add(admin)
            db.session.commit()
           
            # 在執行緒中啟動Flask伺服器
            # 雖然啟動伺服器的app.run()已經換成flask run命令，但仍然可以使用app.run()
            cls.server_thread = threading.Thread(target=cls.app.run, kwargs={"debug": "false", "use_reloader": False, "use_debugger": False})
            cls.server_thread.start()
            # 給伺服器一點時間以確保其正常運作
            time.sleep(1)
   
    @classmethod
    def tearDownClass(cls):
        if cls.client:
            # 停止Flask伺服器與瀏覽器
            cls.client.get("http://localhost:5000/shutdown")
            cls.client.quit()
            # 將主執行緒暫停，等待指定的執行緒結束
            cls.server_thread.join()
           
            # 銷毀資料庫
            db.drop_all()
            db.session.remove()
           
            # 移除app context
            cls.app_context.pop()
   
    def setUp(self):
        if not self.client:
            self.skipTest("Web browser not available")
   
    def tearDown(self):
        pass
   
    # Selenium單元測試範例
    def test_admin_home_page(self):
        # 前往首頁
        self.client.get("http://localhost:5000/")
        self.assertTrue(re.search("Hello,\s+Stranger!", self.client.page_source))
       
        # 前往登入頁
        self.client.find_element_by_link_text("Log In").click()
        self.assertIn("<h1>Login</h1>", self.client.page_source)
       
        # 登入
        self.client.find_element_by_name("email").send_keys("john@example.com")
        self.client.find_element_by_name("password").send_keys("cat")
        self.client.find_element_by_name("submit").click()
        self.assertTrue(re.search("Hello,\s+john!", self.client.page_source)
       
        # 前往使用者的個人資訊網頁
        self.client.find_element_by_link_text("Profile").click()
        self.assertIn("<h1>john</h1>", self.client.page_source)

第十六章、效能－分析與記錄緩慢的資料庫查詢(16a、16b)

l   分析與記錄緩慢的資料庫查詢：
git checkout 16a
git checkout 16b

l   app/main/views.py：
# Flask-SQLAlchemy有個選項可記錄在處理一個request的過程中發出的資料庫查詢統計數據，回報緩慢的資料庫查詢
# ...
from flask_sqlalchemy import get_debug_queries
# ...

@main.after_app_request
def after_request(response):
    # 在預設情況下，get_debug_queries()函式只會在除錯模式中啟用
    for query in get_debug_queries():
        if query.duration >= current_app.config["FLASKY_SLOW_DB_QUERY_TIME"]:
            current_app.logger.warning("Slow query: %s\nParameters: %s\nDuration: %fs\nContext: %s\n" % (query.statement, query.parameters, query.duration, query.context))
    return response
# ...

l   config.py：
# ...
class Config:
    # ...
    FLASKY_SLOW_DB_QUERY_TIME = 0.5
    # ...
# ...

l   Flask-SQLAlchemy記錄的查詢統計數據：

statement

SQL陳述式

parameters

SQL陳述式的參數

start_time

查詢被發出時的時間

end_time       

查詢回傳時的時間

duration

查詢的執行秒數

context

指出發出查詢的原始程式碼位置的字串

l   flasky.py：
# Flask的開發web伺服器(來自Werkzeug)可讓你選擇為每個request啟用Python剖析，每當web伺服器指派request給app時，WSGI中介軟體就會執行，並修改request的處理方式
# Python剖析器官方文件https://docs.python.org/2/library/profile.html

# ...
@app.cli.command()
@click.option("--length", default=25, help="Number of functions to include in the profiler report.")
@click.option("--profile-dir", default=None, help="Directory where profiler data files are saved.")
def profile(length, profile_dir):
    """Start the application under the code profiler."""
    from werkzeug.contrib.profiler import ProfilerMiddleware
    app.wsgi_app = ProfilerMiddleware(app.wsgi_app, restrictions=[length], profile_dir=profile_dir)
    app.run(debug=False)

第十七章、部署－部署指令(17a)

l   部署指令：
git checkout 17a

l   flasky.py：
# ...
from flask_migrate import upgrade
from app.models import Role, User
# ...

@app.cli.command()
def deploy():
    """Run deployment tasks."""
    # 將資料庫遷移至最新版本
    upgrade()
   
    # 建立或更新使用者角色
    Role.insert_roles()
   
    # 確保所有使用者都追隨他們自己
    User.add_self_follows()

第十七章、部署－以Email通知應用程式錯誤(17b)

l   以Email通知應用程式錯誤：
git checkout 17b

l   在啟動過程中，Flask會建立Python類別logging.Logger的實例，並將它指派給app實例，成為app.logger，在除錯模式中，這個紀錄器會將訊息寫到主控台，需要加入處理常式，否則Log不會被存起來。

l   config.py：
# ...
# 寄送回報app錯誤的email
class ProductionConfig(Config):
    # ...
    @classmethod
    def init_app(cls, app):
        Config.init_app(app)
        # 將錯誤寄給管理員
        import logging
        from logging.handlers import SMTPHandler
        credentials = None
        secure = None
        # Python的getattr()函式用來取得物件屬性值，三個參數分別是物件、屬性名稱、預設值
        if getattr(cls, "MAIL_USERNAME", None) is not None:
            credentials = (cls.MAIL_USERNAME, cls.MAIL_PASSWORD)
            if getattr(cls, "MAIL_USE_TLS", None):
                secure = ()
        mail_handler = SMTPHandler(
            mailhost=(cls.MAIL_SERVER, cls.MAIL_PORT),
            fromaddr=cls.FLASKY_MAIL_SENDER,
            toaddrs=[cls.FLASKY_ADMIN],
            subject=cls.FLASKY_MAIL_SUBJECT_PREFIX + " Application Error",
            credentials=credentials,
            secure=secure)
        mail_handler.setLevel(logging.ERROR)
        app.logger.addHandler(mail_handler)
# ...

第十七章、部署－Heroku支援Gunicorn與Waitress範例(17c、17c-waitress)

l   Heroku支援Gunicorn與Waitress範例：
git checkout 17c
git checkout 17c-waitress

l   Heroku是最早的PaaS供應商之一，Heroku使用一種名為dynos的計算單位來衡量服務的使用與收費，最常見的dyno類型是web dyno，也就是一個web伺服器實例，另一種dyno是worker dyno，它的用途是執行背景工作或其它支援工作。

l   Heroku部署步驟：
建立Heroku帳號、
安裝Heroku CLI、
建立app(git與config設定)、
配置資料庫、
設置登入(log組態設定)、
設置email、
加入頂層的需求檔

l   Heroku會使用自己的SSL憑證，為了確保app的安全，唯一必要的動作就是攔截送往http://介面的所有請求，並將它們轉址到https://，這正是Flask-SSLify擴充套件的功能：
pip install flask-sslify

l   app/__init__.py：
# ...
# 將所有request轉址到安全HTTP
def create_app(config_name):
    # ...
    if app.config["SSL_REDIRECT"]:
        from flask_sslify import SSLify
        sslify = SSLify(app)
    # ...

l   config.py：
# ...
class Config:
    # ...
    SSL_REDIRECT = False
    # ...
# ...
# 支援代理伺服器
class HerokuConfig(ProductionConfig):
    SSL_REDIRECT = True if os.environ.get("DYNO") else False
   
    @classmethod
    def init_app(cls, app):
        ProductionConfig.init_app(app)
        # 處理反向代理伺服器標頭
        try:
            from werkzeug.middleware.proxy_fix import ProxyFix
        except ImportError:
            from werkzeug.contrib.fixers import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app)
        # 記錄Log至stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)

config = {
    "development": DevelopmentConfig,
    "testing": TestingConfig,
    "production": ProductionConfig,
    "heroku": HerokuConfig,
    "default": DevelopmentConfig
}

l   Gunicorn與uWSGI是可以和Flask app良好配合的兩種準產品web伺服器，要在本地端用Gunicorn運行app的指令，Gunicorn預設使用連接埠8000，而不是像Flask的5000：
pip install gunicorn

# flasky:app引數可讓Gunicorn知道app實例在哪裡
gunicorn flasky:app

l   Gunicorn web伺服器無法在Microsoft Windows上運行，如果想要在Windows系統上測試Heroku部署，可使用Waitress：
pip install waitress

waitress-serve -port 8000 flasky:app

第十七章、部署－Docker(17d、17e、17f)

l   Docker：
git checkout 17d
git checkout 17e
git checkout 17f

l   Dockerfile：
# 容器映像組建腳本
FROM python:3.6-alpine

ENV FLASK_APP flasky.py
ENV FLASK_CONFIG production

# adduser的-D引數會停止使用者密碼的互動提示
RUN adduser -D flasky
USER flasky

WORKDIR /home/flasky

COPY requirements requirements
RUN python -m venv venv
RUN venv/bin/pip install -r requirements/docker.txt

COPY app app
COPY migrations migrations
COPY flasky.py config.py boot.sh ./

# 執行期組態
EXPOSE 5000
ENTRYPOINT ["./boot.sh"]

l   config.py：
# ...
# Docker組態設置
class DockerConfig(ProductionConfig):
    @classmethod
    def init_app(cls, app):
        ProductionConfig.init_app(app)
       
        # log至stderr
        import logging
        from logging import StreamHandler
        file_handler = StreamHandler()
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)

config = {
    # ...
    "docker": DockerConfig,
    # ...
}

l   boot.sh：
# 容器啟動腳本
#!/bin/sh
source venv/bin/activate

while true; do
    flask deploy
    if [[ "$?" == "0" ]]; then
        break
    fi
    echo Deploy command failed, retrying in 5 secs...
    sleep 5
done

exec gunicorn -b :5000 --access-logfile - --error-logfile - flasky:app

l   容器部署指令：
# 部署MySQL 5.7資料庫伺服器，為了能夠連接MySQL資料庫，SQLAlchemy要安裝一個pymysql這類的MySQL用戶端套件
docker run --name mysql -d -e MYSQL_RANDOM_ROOT_PASSWORD=yes -e MYSQL_DATABASE=flasky -e MYSQL_USER=flasky -e MYSQL_PASSWORD=<database-password> mysql/mysql-server:5.7

# 建立容器映像
docker build -t flasky:latest .

# 執行容器
# --link選項可設置新容器與另一個既有容器的連結，冒號分隔來源容器名稱或ID，以及所建立容器的別名
docker run -d -p 8000:5000 --link mysql:dbserver -e DATABASE_URL=mysql+pymysql://flasky:<database-password>@dbserver/flasky -e MAIL_USERNAME=<your-gmail-username> -e MAIL_PASSWORD=<your-gmail-password> flasky:latest

l   docker-compose.yml：
# 使用Docker Composer來調度容器
# Compose會偵測flasky容器的links鍵來找出依賴關係，以正確的順序啟動mysql與flasky容器
version: '3'
services:
  flasky:
    build: .
    ports:
      - "8000:5000"
    env_file: .env
    restart: always
    links:
      - mysql:dbserver
  mysql:
    image: "mysql/mysql-server:5.7"
    env_file: .env-mysql
    restart: always

l   .env檔案裡面有下列變數：
FLASK_APP=flasky.py
FLASK_CONFIG=docker
MAIL_USERNAME=<your-gmail-username>
MAIL_PASSWORD=<your-gmail-password>
DATABASE_URL=mysql+pymysql://flasky:<database-password>@dbserver/flasky

l   .env-mysql檔案裡面有下列變數：
MYSQL_RANDOM_ROOT_PASSWORD=yes
MYSQL_DATABASE=flasky
MYSQL_USER=flasky
MYSQL_PASSWORD=<database-password>

l   Docker Composer部署指令：
docker-compose up -d --build

第十七章、部署－傳統的託管方式(17g)

l   傳統的託管方式：
git checkout 17g

l   使用Python套件輔助將.env檔匯入環境：
pip install python-dotenv

l   flasky.py：
# 從.env檔匯入環境變數
import os
from dotenv import load_dotenv

dotenv_path = os.path.join(os.path.dirname(__file__), ".env")
if os.path.exists(dotenv_path):
    load_dotenv(dotenv_path)
# ...

l   config.py：
# ...
# Unix組態範例
class UnixConfig(ProductionConfig):
    @classmethod
    def init_app(cls, app):
        ProductionConfig.init_app(app)
       
        # log至syslog
        import logging
        from logging.handlers import SysLogHandler
        syslog_handler = SysLogHandler()
        syslog_handler.setLevel(logging.WARNING)
        app.logger.addHandler(syslog_handler)

config = {
    # ...
    "unix": UnixConfig,
    # ...
}

第十八章、其他的資源

l   擴充套件，官方的Flask Extension Registry可尋找其它的擴充套件：
Flask-Babel：支援國際化與當地化
Marshmallow：序列化與解序列化Python物件，適合用來表示API資源
Celery：可處理背景工作的工作佇列
Frozen-Flask：將Flask app轉換成靜態網站
Flask-DebugToolbar：在瀏覽器內的除錯工具
Flask-Assets：合併、縮小與編譯CSS和JavaScript資產
Flask-Session：另一種使用者session的作品，使用伺服器端存儲
Flask-SocketIO：支援WebSocket與長輪詢的Socket.IO伺服器

於
標籤： ,

沒有留言:

張貼留言

訂閱： 張貼留言 (Atom)