您當(dāng)前位置：首頁 > php開源 > 綜合技術(shù) > Flask Web 開發(fā)學(xué)習(xí)稿（三）

Flask Web 開發(fā)學(xué)習(xí)稿（三）

來源：程序員人生發(fā)布時間：2016-06-20 07:47:58 閱讀次數(shù)：2880次

第6章電子郵件

當(dāng)我們需要在特定事件產(chǎn)生時提示用戶，包裝了 smtplib 的 Flask-Mail 擴(kuò)大能更好的和 Flask 集成
安裝 pip install flask-mail
Flask-Mail 連接到 SMTP 服務(wù)器，如果不進(jìn)行配置，F(xiàn)lask-Mail 會連接 localhost 上的端口 25

配置	默許值	說明
MAIL_SERVER	localhost	Email服務(wù)器的ip地址或主機(jī)名
MAIL_PORT	25	Email服務(wù)器端口
MAIL_USE_TLS	False	啟用傳輸層安全協(xié)議
MAIL_USE_SSL	False	啟用安全套接曾協(xié)議
MAIL_USERNAME	None	Email用戶名
MAIL_PASSWORD	None	Email密碼

使用外部 SMTP 服務(wù)器更加方便

from flask.ext.mail import Mail

app.config['MAIL_SERVER'] = 'mail.xxx.com'
app.config['MAIL_PORT'] = '587'
app.config['MAIL_USE_TLS'] = 'True'
app.config['MAIL_USERNAME'] = 'username'
app.config['MAIL_PASSWORD'] = 'pwd'
mail = Mail(app)

將賬戶和密碼寫在程序里太不安全了，為了保護(hù)敏感信息，需要讓腳本從環(huán)境變量中導(dǎo)入這些信息

app.config['MAIL_USERNAME'] = os.environ.get('MALI_USERNAME')
app.config['MAIL_PASSWORD'] = os.environ.get('MALI_PASSWORD')

如何設(shè)置環(huán)境變量呢？

# Linux 或 Mac OS X
export MALI_USERNAME=<YOU_USERNAME>
export MALI_PASSWORD=<YOU_PASSWORD>
# Windows 
set MALI_USERNAME=<YOU_USERNAME>
set MALI_PASSWORD=<YOU_PASSWORD>

在程序中集成發(fā)送電子郵件功能

from flask.ext.mail import Message

app.config['FLASKY_MAIL_SUBJECT_PREFIX'] = '[Flasky]'
app.config['FLASKY_MAIL_SENDER'] = 'Flasky Admin <flasky@example.com>'

def send_email(to, subject, template, **kwargs):
    msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + subject,
                  sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    mail.send(msg)

這兩個程序特定配置項，分別定義了郵件主題的前綴和發(fā)件人的地址
send_email() 函數(shù)的參數(shù)分別為收件人地址，主題，渲染郵件正文的模版和關(guān)鍵字參數(shù)列表
指定模版時不能包括擴(kuò)大名，這樣才能使用兩個模版分別渲染純文本正文和富文本正文
調(diào)用者將關(guān)鍵字參數(shù)傳給 render_template() 函數(shù)以便在模版中使用，進(jìn)而生成電子郵件正文，下面修改視圖函數(shù)

app.config['FLASKY_ADMIN'] = os.environ.get('FLASKY_ADMIN')
#...
@app.route('/', methods=['GET', 'POST'])
def index():
    form = NameForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.name.data).first()
        if user is None:
            user = User(username=form.name.data)
            db.session.add(user)
            session['known'] = False
            if app.config['FLASKY_ADMIN']:
                send_email(app.config['FLASKY_ADMIN'], 'New User',
                           'mail/new_user', user=user)
        else:
            session['known'] = True

        session['name'] = form.name.data
        form.name.data = ''

        return redirect(url_for('index'))
    return render_template('index.html', form=form, name=session.get('name'), known=session.get('known', False))

我們要創(chuàng)建兩個模版文件，分別用于渲染純文本和 HTML 版的郵件正文，這兩個模版文件都保存在 tmplates 文件夾下的 mail 子文件夾中，以便和普通模版辨別開來。電子郵件的模版中要有1個模版參數(shù)是用戶，因此調(diào)用 send_mail() 函數(shù)時要以關(guān)鍵字參數(shù)的情勢傳入用戶
這樣的程序會在發(fā)送郵件的時候造成短暫阻塞，異步發(fā)送電子郵件來消除這類沒必要要的延遲

from threading import Thread

def send_async_email(app, msg):
    with app.app_context():
        mail.send(msg)

def send_email(to, subject, template, **kwargs):
    msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + subject,
                  sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    thr = Thread(target=send_async_email, args=[app, msg])
    thr.start()
    return thr

很多 Flask 擴(kuò)大都假定已存在激活的程序上下文和要求上下文，F(xiàn)lask-Mail 中的 send() 函數(shù)使用 current_app，因此必須激活程序上下文，不過不同線程中履行mail.send()函數(shù)時，程序上下文要使用 app.app_context() 人工創(chuàng)建
當(dāng)你需要大量發(fā)送電子郵件時，使用 Celery 任務(wù)隊列更適合

第7章大型程序的結(jié)構(gòu)

現(xiàn)在我們已完成了很多功能的學(xué)習(xí)，但是隨著程序愈來愈大，我們將學(xué)會如何組織大型程序的結(jié)構(gòu)

7.1 項目結(jié)構(gòu)

├─Flsky
│  │─app  # Flask 程序
│  |  ├─static
│  |  |─templates
│  |  |─main
│  |  |  │-__init__.py
│  │  |  |-errors.py
│  │  |  |-forms.py
│  │  |  |-views.py
│  │  |-__init__.py
│  │  |-email.py
│  │  |-models.py
|  |-migrations  # 數(shù)據(jù)庫遷移腳本
|  |-tests  # 單元測試
|  |  |-__init__.py
|  |  |-test*.py
│  |-config.py  # 貯存配置
│  |-manage.py  # 用于啟動程序和其他的程序任務(wù)
│  |-requirements.txt  # 列出全部依賴包
|  └─ venv # python虛擬環(huán)境

7.2 配置選項

從現(xiàn)在開始我們我們的配置不會再像之前那樣用簡單的字典結(jié)構(gòu)，而是使用配置類

#config.py
import os
basedir = os.path.abspath(os.path.dirname(__file__))

# 基類
class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard to guess string'
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True
    FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]'
    FLASKY_MAIL_SENDER = 'Flasky Admin <flasky@example.com>'
    FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN')

    @staticmethod
    def init_app(app):
        pass

class DevelopmentConfig(Config): DEBUG = True
    MAIL_SERVER = 'smtp.googlemail.com'
    MAIL_PORT = 587
    MAIL_USE_TLS = True
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data-dev.sqlite')

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \ 'sqlite:///' + os.path.join(basedir, 'data-test.sqlite')

class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data.sqlite')

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

配置類可以定義init_app()方法，其參數(shù)是程序?qū)嵗谶@個方法中，可以履行對當(dāng)前環(huán)境的配置初始化，現(xiàn)在基類 Config 中的init_app()方法為空

7.3 程序包

程序包用來保存程序的所有代碼、模版和靜態(tài)文件。

7.3.1 使用程序工廠函數(shù)

把創(chuàng)建進(jìn)程移到可顯式調(diào)用的工廠函數(shù)中，程序的工廠函數(shù)在 app 包的構(gòu)造文件中定義
構(gòu)造文件導(dǎo)入了太多正在使用的 Flask 擴(kuò)大，由于還沒有初始化所需的程序?qū)嵗詻]有初始化擴(kuò)大，創(chuàng)建擴(kuò)大類時沒有向構(gòu)造函數(shù)傳入?yún)?shù)
create_app() 函數(shù)就是程序的工廠函數(shù)，接受1個參數(shù)，是程序使用的配置名
配置類在 config.py 文件中定義，其中保存的配置可使用 Flask app.config 配置對象提供的 from_object() 方法直接導(dǎo)入程序。至于擴(kuò)大對象，則可以通過名字從 config 字典當(dāng)選擇。
程序創(chuàng)建配置好后，就可以初始化擴(kuò)大了，在之前創(chuàng)建的擴(kuò)大對象上調(diào)用 init_app() 可以完成初始化進(jìn)程

# app/__init__.py
from flask import Flask, render_template
from flask.ext.bootstrap import Bootstrap
from flask.ext.mail import Mail
from flask.ext.moment import Moment
from flask.ext.sqlalchemy import SQLAlchemy
from config import config

bootstrap = Bootstrap()
mail = Mail()
moment = Moment()
db = SQLAlchemy()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    bootstrap.init_app(app)
    mail.init_app(app)
    moment.init_app(app)
    db.init_app(app)
    # 附加路由和自定義毛病頁面

    return app

7.3.2 在藍(lán)本中實現(xiàn)程序的功能

在單腳本程序中，程序?qū)嵗嬖谟谌肿饔糜蛑校酚煽梢灾苯邮褂?app.route 修飾器定義。但是現(xiàn)在利用程序?qū)嵗沁\行時創(chuàng)建的，app.route 只在在 create_app() 以后才存在，這時候定義路由就太晚了
在藍(lán)本中注冊的路由都處于休眠狀態(tài)，直到藍(lán)本注冊到程序上后，路由才真正成為程序的1部份。使用位于全局作用域的藍(lán)本時，定義路由的方法幾近和單腳本程序1樣
為了取得最大的靈活性，程序包中創(chuàng)建了1個子包用于保存藍(lán)本

app/main/__init__.py
from flask import Blueprint
main = Blueprint('main', __name__)
from . import views, errors

通過實例化1個藍(lán)本類對象可以創(chuàng)建藍(lán)本，構(gòu)造函數(shù)有兩個參數(shù)：藍(lán)本的名字和藍(lán)本所在的模塊或包，大多數(shù)情況下，Python的 __name__變量就是第2個參數(shù)所需要的值
利用程序的路由放在app/main/views.py模塊中，毛病處理放在app/main/errors.py中。導(dǎo)入這些模塊以后，路由和毛病處理就和藍(lán)本關(guān)聯(lián)起來了。
有1點要注意，路由和毛病處理模塊要在 app/__init__.py 的底部被導(dǎo)入，由于views.py 和 errors.py 要導(dǎo)入 main blueprint，所以為了不循環(huán)依賴我們要等到 main 被創(chuàng)建出來才能夠?qū)肼酚珊兔√幚怼?
藍(lán)本在工廠函數(shù) create_app() 中注冊到程序上

# app/__init__.py 注冊藍(lán)本
def create_app(config_name):
    # ...
    from main import main as main_blueprint
    app.register_blueprint(main_blueprint)
    return app

毛病處理程序以下

#app/main/error.py
from flask import render_template
from . import main

@main.app_errorhandler(404)
def page_not_found(e):
    return render_template('404.html'), 404

@main.app_errorhandler(500)
def internal_server_error(e):
    return render_template('500.html'), 500

如果使用 errorhandler 修飾器，只有藍(lán)本中的毛病才能觸發(fā)處理程序，要想注冊程序全局的毛病處理程序，必須使用 app_errorhandler
在藍(lán)本中定義路由以下

# app/main/views.py
from datetime import datetime
from flask import render_template, session, redirect, url_for
from . import main
from .forms import NameForm
from .. import db
from ..models import User

@main.route('/', methods=['GET', 'POST'])
def index():
    form = NameForm()
    if form.validate_on_submit():
        # ...
        return redirect(url_for('.index'))
    return render_template('index.html',
                           form=form, name=session.get('name'),
                           known=session.get('known', False),
                           current_time=datetime.utcnow())

在藍(lán)本中的視圖函數(shù)的區(qū)分

路由修飾器由藍(lán)本提供
url_for() 的用法不同

Flask 會為藍(lán)本中的全部端點都加上1個命名空間，這樣就能夠在不同的藍(lán)本中使用相同的端點名定義視圖函數(shù)，而不會產(chǎn)生沖突。命名空間就是藍(lán)本的名字（藍(lán)本構(gòu)造函數(shù)的第1個參數(shù)），所以視圖函數(shù) index() 注冊的端點名是 main.inedx 其 URL 使用 url_for('main.index') 獲得
為了完全修改程序的頁面，表單對象也要移到藍(lán)本中，保存于 app/main/forms.py 模塊

7.4 啟動腳本

manage.py 文件用于啟動程序

#!/usr/bin/env python
import os
from app import create_app, db
from app.models import User, Role
from flask.ext.script import Manager, Shell
from flask.ext.migrate import Migrate, MigrateCommand

app = create_app(os.getenv('FLASK_CONFIG') or 'default')
manager = Manager(app)
migrate = Migrate(app, db)

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role)

manager.add_command("shell", Shell(make_context=make_shell_context))
manager.add_command('db', MigrateCommand)

if __name__ == '__main__':
    manager.run()

該腳本首先創(chuàng)建利用程序?qū)嵗缓髲南到y(tǒng)環(huán)境中讀取FLASK_CONFIG變量，如果該變量沒有定義則使用默許值。然后初始化Flask-Script, Flask-Migrate和為 Python Shell 定義的上下文

7.5 需求文件

pip 可使用以下命令自動生成這個文件
pip freeze >requirements.txt
當(dāng)你在另外一個環(huán)境中準(zhǔn)備安裝這些依賴時，履行以下命令
pip install -r requirements.txt

7.6 單元測試

import unittest
from flask import current_app
from app import create_app, db

class BasicsTestCase(unittest.TestCase):

    def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()

    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self.app_context.pop()

    def test_app_exists(self):
        self.assertFalse(current_app is None)

    def test_app_is_testing(self):
        self.assertTrue(current_app.config['TESTING'])

測試是依照典型的單元測試的寫法來構(gòu)建的，類似于運行中的程序，首先使用測試配置創(chuàng)建程序，然后激活上下文。setUp() 和 tearDown() 方法在每一個測試方法履行前后都會運行，任何以test_ 開頭的方法都會被當(dāng)作測試方法來履行。setUp()方法創(chuàng)建了測試所需的環(huán)境，他首先創(chuàng)建了利用程序?qū)嵗米鳒y試的山下文環(huán)境，這樣就可以確保測試拿到current_app, 然后新建了1個全新的數(shù)據(jù)庫。數(shù)據(jù)庫和利用程序?qū)嵗詈蠖紩?code>tearDown() 方法被燒毀。

第1個測試確保了利用程序?qū)嵗谴嬖诘模?個測試?yán)贸绦驅(qū)嵗跍y試配置下運行。為了確保測試文件夾有正確的包結(jié)構(gòu)，我們需要添加1個tests/__init__.py 文件，這樣單元測試包就可以掃描所有在測試文件夾中的模塊了。

為了運行單元測試，我們可以在manage.py中添加1個自定義命令,

@manager.command
def test():
    """Run the unit tests."""
    import unittest
    tests = unittest.TestLoader().discover('tests')
    unittest.TextTestRunner(verbosity=2).run(tests)

運行方法以下

(venv) $ python manage.py test
test_app_exists (test_basics.BasicsTestCase) ... ok
test_app_is_testing (test_basics.BasicsTestCase) ... ok
.----------------------------------------------------------------------
Ran 2 tests in 0.001s
OK