配套視頻教程
本章將使用適合大型應用程序的樣式重新構建應用程序。
目前Microblog已是一個體面的應用程序了,所以此刻是討論一個Flask應用程序如何成長而不會變得混亂或太難管理的時機了。Flask是一個框架,旨在為我們提供以想要的任何方式組織項目的選項,並且作為這個理念的一部分,它可以在應用程序變得更大、或我們的需要以及經驗水平提升時,更改或調整應用程序的結構。
接下來,將討論適用於大型應用程序的一些模式,並演示對Microblog項目的結構進行一些更改,目標是使代碼更易於維護和更有條理。但是當然,在真正的Flask精神中,鼓勵嘗試決定組織自己的項目時,將這些更改作為建議。
目前的限制
應用程序在目前為止存在兩個基本問題。看一下應用程序的結構,會注意到有一些不同的子系統可識別,但支持它們的代碼是混合的,沒有任何明確的界限。回顧一下這些子系統是什么:
- 用戶身份驗證子系統,包括app/routes.py中的一些視圖功能,app/forms.py中的一些表單,app/templates中的一些模板,以及app/email.py中的電子郵件支持。
- 錯誤子系統,app/errors.py中定義的錯誤處理 和app/templates中的模板。
- 核心應用程序功能,包括顯示和編寫博客帖子,用戶個人資料和關注,以及博客帖子的實時翻譯,這些功能通過大多數應用程序的模塊 和模板進行傳播。
考慮已確定的這個三個子系統以及它們的結構,可能會注意到一種模式。到目前為止,一直遵循的組織邏輯是基於具有專用於不同應用程序功能的模塊。有一個視圖函數模塊、一個用於Web表單、一個用於錯誤處理、一個用於電子郵件,一個用於HTML模板,等等。雖然這是一個對於小項目有意義的結構,但是一旦項目開始增長,它往往會使這些模塊中的一些變得非常龐大和混亂。
清楚地看到問題的一種方法是考慮如何通過盡可能多地重用這個項目來啟動第二個項目。例如,用戶身份驗證部分 應該可以在其他應用程序中正常運行,但如果想要按原樣使用這個代碼,則必須進入多個模塊,並將相關部分復制、粘貼到新項目的新文件中。這將是極其不方便的。如果這個項目將所有與身份驗證相關的文件與應用程序的其余部分分開,這就非常好了。Flask藍圖功能有助於實現更實用的組織方式,使重用代碼變得更容易。
第二個問題不明顯。Flask應用程序實例 在app/init.py中以一個全局變量被創建,然后被很多應用程序模塊所導入。雖然這本身不是一個問題,但將應用程序作為全局變量可能會使得某些場景復雜化,特別是與測試相關的場景中。想象一下,若想在不同配置下測試這個應用程序。由於應用程序被定義為全局變量,因此實際上無法實例化使用不同配置變量的兩個應用程序。另一種不理想的情況是 所有測試都使用相同的應用程序,因此測試可能會對應用程序進行更改,從而影響以后運行的另一個測試。理想情況下,希望所有測試都在一個質朴的應用程序實例上運行。
實際上在tests.py模塊中可看到,在應用程序中設置后要求修改配置,以指示測試使用內存數據庫而不是基於磁盤的默認SQLite數據庫。我們真的沒有其他辦法去更改已配置的數據庫,因為在測試開始時,已經創建並配置了應用程序。對於這種特殊情況,在應用程序運用於應用程序后去更改配置似乎工作正常,但在其他情況下可能不會,並且在任何情況下,這都是一種不好的做法,可能導致模糊和難以發現的錯誤。
更好的解決方案是 不讓應用程序使用全局變量,而是使用一個應用程序工廠函數在運行時去創建函數。這將是一個接受配置對象作為參數的函數,並返回一個設置了這些配置的Flask應用程序實例。如果我可以修改應用程序以使用應用程序工廠函數,那么編寫需要特殊配置的測試講變得容易,因為每個測試都可以創建自己的應用程序。
在本章中,我將重構應用程序,為上面提到的三個子系統和應用程序工廠函數引入藍圖。向大家顯示更改的詳細列表是不切實際的,因為屬於應用程序的一部分的每個文件幾乎沒有變化,所以我將討論重構所采取的的步驟。可在源碼文件中查看這些更改。
藍圖
在Flask中,藍圖是表示應用程序子集的邏輯結構。藍圖可包括路由、視圖函數、表單、模板、靜態文件等等元素。如果我們在單獨的Python包中編寫藍圖,那么我們將擁有一個組件,這個組件封裝了與應用程序的特定功能相關的元素。
一個藍圖的內容最初是處於休眠狀態。要關聯這些元素,需要在應用程序中注冊藍圖。在注冊期間,添加到藍圖的所有元素都會傳遞給應用程序。因此,可將藍圖視為應用程序功能的臨時存儲,以幫助組織代碼。
錯誤處理藍圖
創建的第一個藍圖是封裝對錯誤處理的支持的藍圖。這個藍圖結構如下:
app/
errors/ <-- blueprint package
__init__.py <-- blueprint creation
handlers.py <-- error handlers
templates/
errors/ <-- error templates
404.html
500.html
__init__.py <-- blueprint registration
本質上,所做的是將app/errors.py模塊 移動到app/errors/handlers.py中;將這兩個錯誤模板移動到app/templates/errors中,以便它們與其他模板分開。還必須更改在兩個錯誤處理中的render_template()
調用,以使用的新的錯誤模板子目錄。之后,在創建應用程序實例后,將藍圖創建添加到app/errors/init.py模塊中,並將藍圖注冊添加到app/init.py中。
應該注意到,可將Flask藍圖配置為具有模板或靜態文件的單獨目錄。我已決定將模板移動到應用程序模板目錄的子目錄中,以便所有模板都在單個層次結構中,但如果希望在藍圖包中具有屬於藍圖的模板,那是支持的。例如,如果向Blueprint()
構造函數添加一個template_folder='templates'
參數,那么可將藍圖的模板存儲在app/errors/templates中。
創建藍圖與創建應用程序非常相似。這是在藍圖包的__init__.py模塊中完成的:
app/errors/init.py:錯誤藍圖
from flask import Blueprint
bp = Blueprint('errors', __name__)
from app.errors import handlers
Blueprint類 采取藍圖的名字,基礎模塊的名字(如同在Flask應用程序實例中通常設置__name__),以及一些可選參數(在這種情況下,我不需要)。創建藍圖對象后,我導入handlers.py模塊,以便其中的錯誤處理程序注冊到藍圖。這個導入位於底部以避免循環依賴。
在handlers.py模塊中,我沒有使用@app.errorhandler
裝飾器將錯誤處理程序附加到應用程序中,而是使用藍圖的@bp.app_errorhandler
裝飾器。雖然兩個裝飾器都達到了相同的最終結果,但我們的想法是嘗試 使藍圖獨立於應用程序,以便它更具可移植性。還需要修改兩個錯誤模板的路徑,以便考慮移動它們的新錯誤子目錄。
app/errors/handlers.py:
from flask import render_template
from app import db
from app.errors import bp
@bp.app_errorhandler(404)
def not_found_error(error):
return render_template('errors/404.html'), 404
@bp.app_errorhandler(505)
def internal_error(error):
db.session.rollback()
return render_template('errors/500.html'), 500
完成錯誤處理程序重構的最后一步是 在應用程序中注冊藍圖:
app/init.py:在應用程序中注冊錯誤藍圖
#...
def create_app(config_class=Config):
#...
app = Flask(__name__)
# ...
from app.errors import bp as errors_bp
app.register_blueprint(errors_bp)
# ...
from app import routes, models # <-- remove errors from this import!
要注冊藍圖,得使用Flask應用程序實例的register_blueprint()
方法。注冊藍圖時,任何視圖函數、模板、靜態文件、錯誤處理模塊等都會連接到應用程序。我把藍圖的導入放在app.register_blueprint()
的上方,以避免循環依賴。
身份驗證藍圖
將應用程序的身份驗證功能重構為一個藍圖的過程 與上一小節錯誤處理程序的過程非常相似。如下是重構藍圖的結構圖:
<pre><code>
app/
auth/ <-- blueprint package
__init__.py <-- blueprint creation
email.py <-- authentication emails
forms.py <-- authentication forms
routes.py <-- authentication routes
templates/
auth/ <-- blueprint templates
login.html
register.html
reset_password_request.html
reset_password.html
__init__.py <-- blueprint registration
</code></pre>
要創建這個藍圖,必須將所有與身份驗證相關的功能移動到 我在藍圖創建的新模塊中。包括一些視圖函數、Web表單、以及通過電子郵件發送密碼重置令牌的支持函數等。還將模板移動到子目錄中,以將它們與應用程序的其余部分分開。
在一個藍圖中定義路由時,得使用@bp.route
裝飾,而不是@app.route
。在url_for()
去構建URL時,使用的語法也需要做出更改。對於直接附加到應用程序的常規視圖函數,url_for()
的第一個參數是視圖函數的名字。在一個藍圖中當定義一個路由時,這個參數必須包含藍圖名字和視圖函數名字,並以逗號隔開。例如,我必須用url_for('auth.login')
替換所有出現的url_for('login')
,並且對剩下的視圖函數也是如此。
在應用程序中注冊 auth
藍圖,使用了稍有不同的格式:
app/init.py:
# ...
from app.auth import bp as auth_bp
app.register_blueprint(auth_bp, url_prefix='/auth')
# ...
在這種情形下的register_blueprint()
調用有一個額外的參數 url_prefix
。這完全是可選的,但是Flask為我們提供了在URL前綴下附加藍圖的選項,因此任何在藍圖中定義的路由都會在它們的URL中取得前綴。在很多情況下,這可用作一種“命名空間”,它可將藍圖中的所有路由與應用程序或其他藍圖中的其他路由分開。對於身份驗證,我認為讓所有路由以 /auth
開頭很好,所有就添加了此前綴。所以現在登錄URL將是 http://localhost:5000/auth/login
。因為用url_for()
生成URL,所以所有URL都會自動包含前綴。
app/auth/init.py:
from flask import Blueprint
bp = Blueprint('auth', __name__)
from app.auth import routes
app/auth/email.py
from flask import render_template, current_app
from flask_babel import _
from app.email import send_email
def send_password_reset_email(user):
token = user.get_reset_password_token()
send_email(_('[Microblog] Reset Your Password'),
sender=current_app.config['ADMINS'][0],
recipients=[user.email],
text_body=render_template('email/reset_password.txt',
user=user, token=token),
html_body=render_template('email/reset_password.html',
user=user, token=token))
app/auth/forms.py:
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import ValidationError, DataRequired, Email, EqualTo
from flask_babel import _, lazy_gettext as _l
from app.models import User
class LoginForm(FlaskForm):
username = StringField(_l('Username'), validators=[DataRequired()])
password = PasswordField(_l('Password'), validators=[DataRequired()])
remember_me = BooleanField(_l('Remember Me'))
submit = SubmitField(_l('Sign In'))
class RegistrationForm(FlaskForm):
username = StringField(_l('Username'), validators=[DataRequired()])
email = StringField(_l('Email'), validators=[DataRequired(), Email()])
password = PasswordField(_l('Password'), validators=[DataRequired()])
password2 = PasswordField(
_l('Repeat Password'), validators=[DataRequired(),
EqualTo('password')])
submit = SubmitField(_l('Register'))
def validate_username(self, username):
user = User.query.filter_by(username=username.data).first()
if user is not None:
raise ValidationError(_('Please use a different username.'))
def validate_email(self, email):
user = User.query.filter_by(email=email.data).first()
if user is not None:
raise ValidationError(_('Please use a different email address.'))
class ResetPasswordRequestForm(FlaskForm):
email = StringField(_l('Email'), validators=[DataRequired(), Email()])
submit = SubmitField(_l('Request Password Reset'))
class ResetPasswordForm(FlaskForm):
password = PasswordField(_l('Password'), validators=[DataRequired()])
password2 = PasswordField(
_l('Repeat Password'), validators=[DataRequired(),
EqualTo('password')])
submit = SubmitField(_l('Request Password Reset'))
app/auth/routes.py:
from flask import render_template, redirect, url_for, flash, request
from werkzeug.urls import url_parse
from flask_login import login_user, logout_user, current_user
from flask_babel import _
from app import db
from app.auth import bp
from app.auth.forms import LoginForm, RegistrationForm, \
ResetPasswordRequestForm, ResetPasswordForm
from app.models import User
from app.auth.email import send_password_reset_email
@bp.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect(url_for('main.index'))
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(username=form.username.data).first()
if user is None or not user.check_password(form.password.data):
flash(_('Invalid username or password'))
return redirect(url_for('auth.login'))
login_user(user, remember=form.remember_me.data)
next_page = request.args.get('next')
if not next_page or url_parse(next_page).netloc != '':
next_page = url_for('main.index')
return redirect(next_page)
return render_template('auth/login.html', title=_('Sign In'), form=form)
@bp.route('/logout')
def logout():
logout_user()
return redirect(url_for('main.index'))
@bp.route('/register', methods=['GET', 'POST'])
def register():
if current_user.is_authenticated:
return redirect(url_for('main.index'))
form = RegistrationForm()
if form.validate_on_submit():
user = User(username=form.username.data, email=form.email.data)
user.set_password(form.password.data)
db.session.add(user)
db.session.commit()
flash(_('Congratulations, you are now a registered user!'))
return redirect(url_for('auth.login'))
return render_template('auth/register.html', title=_('Register'),
form=form)
@bp.route('/reset_password_request', methods=['GET', 'POST'])
def reset_password_request():
if current_user.is_authenticated:
return redirect(url_for('main.index'))
form = ResetPasswordRequestForm()
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
if user:
send_password_reset_email(user)
flash(
_('Check your email for the instructions to reset your password'))
return redirect(url_for('auth.login'))
return render_template('auth/reset_password_request.html',
title=_('Reset Password'), form=form)
@bp.route('/reset_password/<token>', methods=['GET', 'POST'])
def reset_password(token):
if current_user.is_authenticated:
return redirect(url_for('main.index'))
user = User.verify_reset_password_token(token)
if not user:
return redirect(url_for('main.index'))
form = ResetPasswordForm()
if form.validate_on_submit():
user.set_password(form.password.data)
db.session.commit()
flash(_('Your password has been reset.'))
return redirect(url_for('auth.login'))
return render_template('auth/reset_password.html', form=form)
以下是模板:
app/templates/auth/login.html:
{% extends "base.html" %}
{% import 'bootstrap/wtf.html' as wtf %}
{% block app_content %}
<h1>{{ _('Sign In') }}</h1>
<div class="row">
<div class="col-md-4">
{{ wtf.quick_form(form) }}
</div>
</div>
<br>
<p>{{ _('New User?') }} <a href="{{ url_for('auth.register') }}">{{ _('Click to Register!') }}</a></p>
<p>
{{ _('Forgot Your Password?') }}
<a href="{{ url_for('auth.reset_password_request') }}">{{ _('Click to Reset It') }}</a>
</p>
{% endblock %}
app/templates/auth/register.html:
{% extends "base.html" %}
{% import 'bootstrap/wtf.html' as wtf %}
{% block app_content %}
<h1>{{ _('Register') }}</h1>
<div class="row">
<div class="col-md-4">
{{ wtf.quick_form(form) }}
</div>
</div>
{% endblock %}
app/templates/auth/reset_password.html:
{% extends "base.html" %}
{% import 'bootstrap/wtf.html' as wtf %}
{% block app_content %}
<h1>{{ _('Reset Your Password') }}</h1>
<div class="row">
<div class="col-md-4">
{{ wtf.quick_form(form) }}
</div>
</div>
{% endblock %}
app/templates/auth/reset_password_request.html:
{% extends "base.html" %}
{% import 'bootstrap/wtf.html' as wtf %}
{% block app_content %}
<h1>{{ _('Reset Password') }}</h1>
<div class="row">
<div class="col-md-4">
{{ wtf.quick_form(form) }}
</div>
</div>
{% endblock %}
應用程序【主main】藍圖
第三個藍圖 包含核心應用程序的邏輯。重構這個藍圖需要使用與前兩個藍圖相同的過程。給這個藍圖取名為main
,所以引用視圖函數的所有url_for()
調用都必須得到一個main.
的前綴。鑒於這是應用程序的核心功能,我決定將模板保留在相同的位置。
app/main/init.py:
from flask import Blueprint
bp = Blueprint('main', __name__)
from app.main import routes
app/main/forms.py:
from flask import request
from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField, TextAreaField
from wtforms.validators import ValidationError, DataRequired, Length
from flask_babel import _, lazy_gettext as _l
from app.models import User
class EditProfileForm(FlaskForm):
username = StringField(_l('Username'), validators=[DataRequired()])
about_me = TextAreaField(_l('About me'),
validators=[Length(min=0, max=140)])
submit = SubmitField(_l('Submit'))
def __init__(self, original_username, *args, **kwargs):
super(EditProfileForm, self).__init__(*args, **kwargs)
self.original_username = original_username
def validate_username(self, username):
if username.data != self.original_username:
user = User.query.filter_by(username=self.username.data).first()
if user is not None:
raise ValidationError(_('Please use a different username.'))
class PostForm(FlaskForm):
post = TextAreaField(_l('Say something'), validators=[DataRequired()])
submit = SubmitField(_l('Submit'))
app/main/routes.py:
from datetime import datetime
from flask import render_template, flash, redirect, url_for, request, g, \
jsonify, current_app
from flask_login import current_user, login_required
from flask_babel import _, get_locale
from guess_language import guess_language
from app import db
from app.main.forms import EditProfileForm, PostForm
from app.models import User, Post
from app.translate import translate
from app.main import bp
@bp.before_app_request
def before_request():
if current_user.is_authenticated:
current_user.last_seen = datetime.utcnow()
db.session.commit()
#g.locale = str(get_locale())
g.locale = 'zh' if str(get_locale()).startswith('zh') else str(get_locale())
@bp.route('/', methods=['GET', 'POST'])
@bp.route('/index', methods=['GET', 'POST'])
@login_required
def index():
form = PostForm()
if form.validate_on_submit():
language = guess_language(form.post.data)
if language == 'UNKNOWN' or len(language) > 5:
language = ''
post = Post(body=form.post.data, author=current_user,
language=language)
db.session.add(post)
db.session.commit()
flash(_('Your post is now live!'))
return redirect(url_for('main.index'))
page = request.args.get('page', 1, type=int)
posts = current_user.followed_posts().paginate(
page, current_app.config['POSTS_PER_PAGE'], False)
next_url = url_for('main.index', page=posts.next_num) \
if posts.has_next else None
prev_url = url_for('main.index', page=posts.prev_num) \
if posts.has_prev else None
return render_template('index.html', title=_('Home'), form=form,
posts=posts.items, next_url=next_url,
prev_url=prev_url)
@bp.route('/explore')
@login_required
def explore():
page = request.args.get('page', 1, type=int)
posts = Post.query.order_by(Post.timestamp.desc()).paginate(
page, current_app.config['POSTS_PER_PAGE'], False)
next_url = url_for('main.explore', page=posts.next_num) \
if posts.has_next else None
prev_url = url_for('main.explore', page=posts.prev_num) \
if posts.has_prev else None
return render_template('index.html', title=_('Explore'),
posts=posts.items, next_url=next_url,
prev_url=prev_url)
@bp.route('/user/<username>')
@login_required
def user(username):
user = User.query.filter_by(username=username).first_or_404()
page = request.args.get('page', 1, type=int)
posts = user.posts.order_by(Post.timestamp.desc()).paginate(page, current_app.config['POSTS_PER_PAGE'], False)
next_url = url_for('main.user', username=user.username, page=posts.next_num) if posts.has_next else None
prev_url = url_for('main.user', username=user.username, page=posts.prev_num) if posts.has_prev else None
return render_template('user.html', user=user, posts=posts.items, next_url=next_url, prev_url=prev_url)
@bp.route('/edit_profile',methods=['GET','POST'])
@login_required
def edit_profile():
form = EditProfileForm(current_user.username)
if form.validate_on_submit():
current_user.username = form.username.data
current_user.about_me = form.about_me.data
db.session.commit()
flash(_('Your changes have been saved.'))
return redirect(url_for('main.edit_profile'))
elif request.method == 'GET':
form.username.data = current_user.username
form.about_me.data = current_user.about_me
return render_template('edit_profile.html', title=_('Edit Profile'), form=form)
@bp.route('/follow/<username>')
@login_required
def follow(username):
user = User.query.filter_by(username=username).first()
if user is None:
flash(_('User %(username)s not found.', username=username))
return redirect(url_for('main.index'))
if user == current_user:
flash(_('You connot follow yourself!'))
return redirect(url_for('main.user', username=username))
current_user.follow(user)
db.session.commit()
flash(_('You are following %(username)s!', username=username))
return redirect(url_for('main.user', username=username))
@bp.route('/unfollow/<username>')
@login_required
def unfollow(username):
user = User.query.filter_by(username=username).first()
if user is None:
flash(_('User %(username)s not found.', username=username))
return redirect(url_for('main.index'))
if user == current_user:
flash(_('You cannot unfollow yourself!'))
return redirect(url_for('main.user', username=username))
current_user.unfollow(user)
db.session.commit()
flash(_('You are not following %(username)s.', username=username))
return redirect(url_for('main.user', username=username))
@bp.route('/translate', methods=['POST'])
@login_required
def translate_text():
return jsonify(
{'text': translate(request.form['text'], request.form['source_language'], request.form['dest_language'])})
應用工廠模式
正如在本章的介紹中提到的那樣,將應用程序作為一個全局變量引入了一些復雜性,主要是對某些測試場景的限制形式。在引入藍圖之前,應用程序必須是一個全局變量,因為所有的視圖函數 和錯誤處理程序都需要用來自app
的裝飾器進行裝飾,例如@app.route
。但是現在所有路由和錯誤處理程序都被移到了藍圖上,使應用程序保持全局的原因要少得多。
因此,我要做的是添加一個名為create_app()
的函數,它構造了Flask應用程序實例,並消除全局變量。轉換不是微不足道的,我不得不不解決一些復雜問題,但讓我們首先看一下應用程序工廠函數:
app/init.py:應用程序工廠函數
import logging
from logging.handlers import SMTPHandler, RotatingFileHandler
import os
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy # 從包中導入類
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_mail import Mail
from flask_bootstrap import Bootstrap
from flask_moment import Moment
from flask_babel import Babel, lazy_gettext as _l
from config import Config # 從config模塊導入Config類
from flask import current_app
db = SQLAlchemy()
migrate = Migrate()
login = LoginManager() # 初始化Flask-Login
login.login_view = 'auth.login'
login.login_message = _l('Please log in to access this page.')
mail = Mail()
bootstrap = Bootstrap()
moment = Moment()
babel = Babel()
def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
db.init_app(app)
migrate.init_app(app, db)
login.init_app(app)
mail.init_app(app)
bootstrap.init_app(app)
moment.init_app(app)
babel.init_app(app)
from app.errors import bp as errors_bp
app.register_blueprint(errors_bp)
from app.auth import bp as auth_bp
app.register_blueprint(auth_bp, url_prefix='/auth')
from app.main import bp as main_bp
app.register_blueprint(main_bp)
if not app.debug and not app.testing:
if not os.path.exists('logs'):
os.mkdir('logs')
file_handler = RotatingFileHandler('logs/microblog.log', maxBytes=10240, backupCount=10)
file_handler.setFormatter(
logging.Formatter('%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Microblog startup')
return app
@babel.localeselector
def get_locale():
return request.accept_languages.best_match(current_app.config['LANGUAGES'])
from app import models # 在此移除errors、routes的導入
如上可看到,大多數Flask擴展都是通過創建擴展的實例並將應用程序作為參數傳遞來初始化的。當應用程序不作為全局變量存在時,有一種替代模式,其中擴展在兩個階段中初始化。擴展實例首先在全局范圍內創建,但不會傳遞任何參數。這將創建未附加到應用程序的擴展實例。在工廠函數中創建應用程序實例時,必須在擴展實例上調用init_app()
方法以將其綁定到現在已知的應用程序。
在初始化期間執行的其他任務保持不變,但是移動到工廠函數 而不是在全局范圍內。包括藍圖注冊和日志配置。請注意,我向條件中添加了一個not app.testing
子句,用於決定是否應該啟用電子郵件和文件日志記錄,以便在單元測試期間跳過所有這些日志記錄。由於在配置中設置了TESTING
變量為True
,因此在運行單元測試時app.testing
標志將為True
。
那么誰調用了應用程序工程函數?使用這個函數的顯而易見的地方是頂級目錄下的microblog.py
腳本,它是應用程序現在在全局范圍內存在的唯一模塊。另一個地方是tests.py
,將在下一節中更詳細地討論單元測試。
正如上面提到的,大多數引用app
隨着藍圖的引入而消失,但是我們必須解決一些代碼中的問題。例如,app/models.py,app/translate.py和app/main/routes.py模塊都有引用app.config
。幸運的是,Flask開發人員試圖讓視圖函數輕松訪問應用程序實例,而不必像以前那樣導入它。Flask提供的current_app
變量是一個特殊的“上下文”變量,Flask在調度請求之前使用應用程序初始化這個變量。在之前已經看過另一個上下文變量,即我正在存儲當前區域設置的g
變量。這兩個,以及Flask-Login的current_user
,還有一些我們沒看到的,有點“神奇”的變量。因為它們像全局變量一樣工作,但只能在處理請求時訪問,並且只能在處理它的線程中訪問。
替換Flask的current_app
變量的app
消除了將應用程序實例作為全局變量導入的需要。通過簡單的搜索和替換,沒有任何困難,我能改變所有引用app.config
和current_app.config
。
app/email.py模塊呈現一個稍大的挑戰,所以我不得不用一個小竅門:
app/email.py:將應用程序實例傳遞給另一個線程
from threading import Thread
from flask import current_app
from flask_mail import Message
from app import mail
def send_async_email(app, msg):
with app.app_context():
mail.send(msg)
def send_email(subject, sender, recipients, text_body, html_body):
msg = Message(subject, sender=sender, recipients=recipients)
msg.body = text_body
msg.html = html_body
Thread(target=send_async_email,args=(current_app._get_current_object(), msg)).start()
在send_email()
函數中,應用程序實例作為參數傳遞給后台線程,然后后台線程將在不阻止應用程序的情況下傳遞電子郵件。直接在send_async_email()
作為后台線程運行的函數中使用current_app
將不起作用,因為current_app
是一個與處理客戶機請求的線程相關聯的上下文感知變量。在不同的線程中,current_app
不會分配值。直接將current_app
作為參數傳遞給線程也不會有效,因為current_app
實際上是一個動態映射到應用程序實例的代理對象。因此傳遞代理對象與使用current_app
相同直接在線程中。我們需要做的是訪問存儲在代理對象中的真實應用程序實例,並將其作為app
參數傳遞。current_app._get_current_object()
表達式從代理對象中提取的實際應用程序實例,所以這就是我傳遞給線程作為參數。
當然還得修改app/templates/email/的reset_password.text和reset_password.html文件中的url_for('auth.reset_password',token=token,_external=True)
添加前綴auth.
。
另一棘手的模塊是 app/cli.py,它實現了一些用於管理語言翻譯的快捷命令。在這種情況下,current_app
變量不起作用,因為這些命令是在啟動時注冊的,而不是在處理請求期間注冊的,這是唯一當current_app
能夠使用的時間。為了刪除在這個模塊中引用向app
,我采用了一個技巧,即 將這些自定義命令移動以app
實例作為參數的register()
函數中:
app/cli.py:注冊自定義應用程序命令
import os
import click
def register(app):
@app.cli.group()
def translate():
#翻譯和本地化命令
pass
@translate.command()
@click.argument('lang')
def init(lang):
#初始化一個新語言
if os.system('pybabel extract -F babel.cfg -k _l -o messages.pot .'):
raise RuntimeError('extract command failed')
if os.system('pybabel init -i messages.pot -d app/translations -l ' +lang):
raise RuntimeError('init command failed')
os.remove('messages.pot')
@translate.command()
def update():
#更新所有語言
if os.system('pybabel extract -F babel.cfg -k _l -o messages.pot .'):
raise RuntimeError('extract command failed')
if os.system('pybabel update -i messages.pot -d app/translations'):
raise RuntimeError('update command failed')
os.remove('messages.pot')
@translate.command()
def compile():
#編譯所有語言
if os.system('pybabel compile -d app/translations'):
raise RuntimeError('compile command failed')
然后從microblog.py中調用這個register()
函數。這是在所有重構后的完整microblog.py:
microblog/microblog.py:重構應用程序模塊
from app import create_app, db, cli
from app.models import User,Post
app = create_app()
cli.register(app)
@app.shell_context_processor
def make_shell_context():
return {'db':db, 'User':User, 'Post':Post}
if __name__ == '__main__':
app.run()
最后,app/models.py保持不變動。
單元測試改進
正如本章開頭所暗示的那樣,到目前為止,所做的很多工作都是為了改進 單元測試工作流程。在運行單元測試時,希望確保應用程序的配置方式不會干擾開發資源,例如 數據庫。
當前版本的tests.py在將應用程序運用於應用程序實例之后 修改配置,這是一種危險的做法,因為並非所有類型的更改都可以在最后完成時運行。我想要的是 有機會在將測試配置添加到應用程序之前指定它。
create_app()
函數現在接受一個配置類作為一個參數。默認情況下,使用在config.py中定義的Config
類,但我現在可以創建一個使用不同配置的應用程序實例,只需將新類傳遞給工廠函數即可。這是一個適合用於我們的單元測試實例配置類:
#...
from app.models import User,Post
from config import Config
class TestConfig(config):
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite://'
#...
上述代碼中 創建了應用程序的Config
類的子類,並重寫SQLAlchemy配置以使用內存中的SQLite數據庫。還添加了一個TESTING
屬性集為True
,目前不需要它,但如果應用程序需要確定它是否在單元測試下運行,則可能會很有用。
應該還記得,我們的單元測試依賴於單元測試框架自動調用的setUp()
和tearDown()
方法來創建和銷毀適用於每個運行測試的環境。現在還可以使用這兩個方法為每個測試創建和銷毀一個全新的應用程序:
microblog/tests.py:為每個測試創建一個應用程序
#...
class UserModelCase(unittest.TestCase):
def setUp(self):
self.app = create_app(TestConfig)
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()
#...
新應用程序將存儲在 self.app
中,但創建應用程序不足以使一切正常。得考慮到db.create_all()
創建數據庫表的語句。db實例需要知道應用程序實例是什么,因為它需要從app.config
中取得數據庫URI,但是當有一個應用程序的工廠,是不是真的局限於單一的應用程序時,可能有一個以上的創建。那么db
如何知道使用我們剛剛創建的 self.app
實例?
答案就在應用程序上下文中。還記得 current_app
變量,當沒有要導入的全局應用程序時,這個變量以某種方式充當應用程序的代理嗎?這個變量在當前線程中查找活動應用程序上下文,如果找到,則從中獲取應用程序。如果沒有上下文,則無法知道哪個應用程序處於活動狀態,因此 current_app
會引發異常。我們可以在下方看到 它在Python控制台中的工作原理。這需要是通過運行 python啟動的控制台,因為flask shell命令會自動激活一個應用程序上下文以方便使用。
(venv) D:\microblog>python
Python 3.6.3 (v3.6.3:2c5fed8, Oct 3 2017, 18:11:49) [MSC v.1900 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> from flask import current_app
>>> current_app.config['SQLALCHEMY_DATABASE_URI']
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "D:\microblog\venv\lib\site-packages\werkzeug\local.py", line 347, in __getattr__
return getattr(self._get_current_object(), name)
File "D:\microblog\venv\lib\site-packages\werkzeug\local.py", line 306, in _get_current_object
return self.__local()
File "D:\microblog\venv\lib\site-packages\flask\globals.py", line 51, in _find_app
raise RuntimeError(_app_ctx_err_msg)
RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
to interface with the current application object in some way. To solve
this, set up an application context with app.app_context(). See the
documentation for more information.
>>> from app import create_app
>>> app = create_app()
[2018-08-29 14:42:53,590] INFO in __init__: Microblog startup
>>> app.app_context().push()
>>> current_app.config['SQLALCHEMY_DATABASE_URI']
'sqlite:///D:\\microblog\\app.db'
這就是秘密!調用視圖函數之前,Flask推送一個應用程序上下文,這使current_app
和 g
處於活動。請求完成后,將刪除上下文、以及這些變量。對於在單元測試setUp()
方法中的db.create_all()
調用,我推送了剛剛創建的應用程序實例的應用程序上下文,並以這種方式,db.create_all()
可以使用current_app.config
來知道數據庫的位置。然后,在tearDown()
方法中彈出上下文將所有內容重置為干凈狀態。
應該還知道 應用程序上下文 是Flask使用的兩個上下文之一。還有一個 請求上下文,它更具體,因為它適用於請求。在處理請求之前激活請求上下文時,Flask的request
和session
變量、以及Flask-Login的current_user
都將變得可用。
環境變量
正如在構建這個應用程序時看到的那樣,有許多配置選項依賴於在啟動服務器之前在我們的環境中設置變量。包括 密鑰、電子郵件服務器信息、數據庫URL、百度翻譯API密鑰等。這樣做很不方便,因為每次打開新的終端會話時,都需要再次設置這些變量。
依賴於大量環境變量的應用程序的常見模式是將它們存儲在根應用程序目錄中的.env
文件中。應用程序會啟動時就會導入此文件中的變量,這樣就不需要手動設置所有這些變量。
有一個支持.env文件的Python包,名為 python-dotenv。
由於config.py
模塊 是我們讀取所有環境變量的地方,因此我將在創建 Config
類之前導入.env
文件,以便在構造類時設置變量:
microblog/config.py:
import os
from dotenv import load_dotenv
basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'))
class Config(object):
# ...
因此,現在可創建一個包含應用程序所需的所有環境變量的 .env
文件。不過請不要將.env
文件添加到 源代碼管理中,因為肯定不希望文件包含源代碼存儲庫包含的密碼和其他敏感信息。
這個.env
文件可用於所有的配置時變量,但是它不能用於Flask的 FLASK_APP
和FLASK_DEBUG
公共環境變量(可使用.flaskenv
文件,參考文檔),因為這些在應用程序的引導過程中是非常早期就需要,即在應用程序實例和其配置對象存在之前。
以下示例顯示了microblog/microblog.env文件:
MAIL_SERVER=smtp.163.com
MAIL_PORT=25
MAIL_USE_TLS=True
MAIL_USE_SSL=False
MAIL_USERNAME=xxxx@163.com
MAIL_PASSWORD=授權碼
APPID=you-baidufanyi-appid
BD_TRANSLATOR_KEY=you-baidufanyi-secret
依賴文件
到目前為止,已在Python虛擬環境中安裝了相當多的軟件包。如果需要在另一台機器上重新生成環境,那么將無法記得必須安裝的軟件包,因此普遍接受的做法是在項目根文件夾中編寫一個requirements.txt
文件,列出所有的依賴項、以及其版本。生成這個列表實際很簡單:
(venv) D:\microblog>pip freeze > requirements.txt
pip freeze
命令 將把安裝在了正確格式的虛擬環境中的所有軟件包存儲在requirements.txt
文件中。現在,如果需要在另一台計算機上創建相同的虛擬環境,而不用逐個安裝軟件包,那么可運行:
pip install -r requirements.txt
目前為止,項目結構
microblog/
app/
auth/
errors/
main/
static/
templates/
auth/
email/
_post.html
base.html
edit_profile.html
index.html
user.html
__init__.py
cli.py
email.py
models.py
translate.py
logs/
migrations/
env/
app.db
babel.cfg
config.py
microblgo.env
microblog.py
requirements.txt
tests.py
參考
https://blog.miguelgrinberg.com/post/the-flask-mega-tutorial-part-xv-a-better-application-structure