[ Python ] Flask 基于 Web开发大型程序的结构实例解析

　　作为一个编程入门新手，Flask是我接触到的第一个Web框架。想要深入学习，就从《FlaskWeb开发：基于Python的Web应用开发实战》这本书入手，本书由于是翻译过来的中文版，理解起来不是很顺畅。但是对着代码理解也是能应对的，学到第七章：大型程序结构这章节的时候，发现难度有所提升，网上能参考的完整实例没有，于是根据自己的理解记下来。

程序结构图：

README

（1）本程序是基于Flask微型Web框架开发，使用Jinja2模版引擎
（2）页面展示了一个文本框和一个按钮，输入文本框点击按钮提交，文本框为空无法提交（输入文本框的数据为一个模拟用户）；
（3）当在文本框中输入新用户提交，欢迎词和文本框中输入老用户提交不一致；
（4）文本框输入新用户提交后，将新用户保存至SQLite数据库，并使用异步发送邮件至管理员邮箱；
（5）页面刷新，浏览器不会再次提示：是否提交


项目结构

flasky     # 程序根目录
├── app     # 核心模块目录
│   ├── email.py     # 邮件发送模版
│   ├── __init__.py
│   ├── main     # 蓝图模块目录
│   │   ├── errors.py     # 错误处理模块
│   │   ├── forms.py     # 页面表单模块
│   │   ├── __init__.py
│   │   └── views.py     # 正常处理模块
│   ├── models.py     # 对象关系映射模块
│   ├── static     # 页面静态资源目录
│   │   └── favicon.ico     # 页面收藏夹图标
│   └── templates     # 默认存放页面模版目录
│       ├── 404.html
│       ├── base.html
│       ├── index.html
│       ├── mail     # 邮件模块目录
│       │   ├── new_user.html
│       │   └── new_user.txt
│       └── user.html
├── config.py     # 程序配置文件
├── data-dev.sqlite     # 程序数据库文件
├── manage.py     # 程序管理启动文件
├── migrations     # 数据库迁移目录
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
├── requirements.txt     # 所有依赖包文件
└── tests     # 测试文件目录
    ├── __init__.py
    └── test_basics.py

README

程序代码总汇

"/"

# -*- coding: utf-8 -*-
# Author: hkey
import os

basedir = os.path.abspath(os.path.dirname(__file__))

class Config(object):   # 所有配置类的父类，通用的配置写在这里
    SECRET_KEY = os.environ.get(\'SECRET_KEY\') or \'hard to guess string\'
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    FLASKY_MAIL_SUBJECT_PREFIX = \'[Flasky]\'
    FLASKY_MAIL_SENDER = \'Flasky Admin <xxx@126.com>\'
    FLASKY_ADMIN = \'xxx@qq.com\'

    @staticmethod
    def init_app(app):  # 静态方法作为配置的统一接口，暂时为空
        pass

class DevelopmentConfig(Config):    # 开发环境配置类
    DEBUG = True
    MAIL_SERVER = \'smtp.126.com\'
    MAIL_PORT = 465
    MAIL_USE_SSL = True
    MAIL_USERNAME = \'xxx@126.com\'
    MAIL_PASSWORD = \'xxxxxx\'
    SQLALCHEMY_DATABASE_URI = \
    \'sqlite:///\' + os.path.join(basedir, \'data-dev.sqlite\')

class TestingConfig(Config):    # 测试环境配置类
    TESTING = True
    SQLALCHEMY_DATABASE_URI = \
        \'sqlite:///\' + os.path.join(basedir, \'data-test.sqlite\')

class ProductionConfig(Config):     # 生产环境配置类
    SQLALCHEMY_DATABASE_URI = \
        \'sqlite:///\' + os.path.join(basedir, \'data.sqlite\')

config = {  # config字典注册了不同的配置，默认配置为开发环境，本例使用开发环境
    \'development\': DevelopmentConfig,
    \'testing\': TestingConfig,
    \'production\': ProductionConfig,
    \'default\': DevelopmentConfig
}

config.py

# -*- coding: utf-8 -*-
# Author: hkey
import os
from app import create_app, db
from app.models import User, Role
from flask_script import Manager, Shell
from flask_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)

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

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

manage.py

"/app"

# -*- coding: utf-8 -*-
# Author: hkey
from flask import Flask, render_template
from flask_bootstrap import Bootstrap
from flask_sqlalchemy import SQLAlchemy
from flask_mail import Mail
from config import config

# 由于尚未初始化所需的程序实例，所以没有初始化扩展，创建扩展类时没有向构造函数传入参数。
bootstrap = Bootstrap()
mail = Mail()
db = SQLAlchemy()

def create_app(config_name):
    \'\'\'工厂函数\'\'\'
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)   # 通过config.py统一接口
    bootstrap.init_app(app)     # 该init_app是bootstrap实例的方法调用，与上面毫无关系
    mail.init_app(app)          # 同上
    db.init_app(app)            # 同上

    # 附加路由和自定义错误页面，将蓝本注册到工厂函数
    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)

    return app

app/__init__.py

# -*- coding: utf-8 -*-
# Author: hkey
from threading import Thread
from flask import render_template, current_app
from flask_mail import Message
from . import mail

def send_async_mail(app, msg):
    \'\'\'创建邮件发送函数\'\'\'
    with app.app_context():
        mail.send(msg)

def send_mail(to, subject, template, **kwargs):
    app = current_app._get_current_object()
    if app.config[\'FLASKY_ADMIN\']:
        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_mail, args=(app, msg))
        thr.start()     # 通过创建子线程实现异步发送邮件
        return thr

app/email.py

# -*- coding: utf-8 -*-
# Author: hkey

# 对象关系映射类

from . import db

class Role(db.Model):
    __tablename__ = \'roles\'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, index=True)
    users = db.relationship(\'User\', backref=\'role\', lazy=\'dynamic\')
    def __repr__(self):
        return \'<Role %r>\' % self.name

class User(db.Model):
    __tablename__ = \'users\'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey(\'roles.id\'))

    def __repr__(self):
        return \'<User %r>\' % self.username

app/models.py

\'\'/app/main"

# -*- coding: utf-8 -*-
# Author: hkey
from  flask import Blueprint
# 定义蓝本
main = Blueprint(\'main\', __name__)

from . import views, errors

app/main/__init__.py

# -*- coding: utf-8 -*-
# Author: hkey
from flask import render_template
from . import main

@main.app_errorhandler(404)     # 路由装饰器由蓝本提供，这里要调用 app_errorhandler 而不是 errorhandler
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

app/main/errors.py

# -*- coding: utf-8 -*-
# Author: hkey
from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import Required

class NameForm(FlaskForm):
    \'\'\'通过 flask-wtf 定义表单类\'\'\'
    name = StringField(\'What is your name ?\', validators=[Required()])  # 文本框
    submit = SubmitField(\'Submit\')  # 按钮

app/main/forms.py

# -*- coding: utf-8 -*-
# Author: hkey
from flask import render_template, session, redirect, url_for, current_app
from . import main
from .forms import NameForm
from .. import db
from ..models import User
from ..email import send_mail

@main.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    # 通过session保存 known为False，通过web渲染需要
            if current_app.config[\'FLASKY_ADMIN\']:  # 如果配置变量有flasky管理员就发送邮件
                # 异步发送邮件
                send_mail(current_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\'))  # 通过redirect避免用户刷新重复提交
    return render_template(\'index.html\', form=form, name=session.get(\'name\'),
                           known=session.get(\'known\', False))

app/main/views.py

"/app/main/templates" 页面

<!DOCTYPE html>
{% extends "bootstrap/base.html" %}
{% block title %}Flasky{% endblock %}
{% block head %}
{{ super() }}
<link rel="shortcut icon" href="{{ url_for(\'static\', filename = \'favicon.ico\')}}"
      type="image/x-icon">
<link rel="icon" href="{{ url_for(\'static\', filename = \'favicon.ico\')}}"
      type="image/x-icon">
{% endblock %}
{% block navbar %}
<div class="navbar navbar-inverse" role="navigation">
    <div class="container">
        <div class="navbar-header">
            <button type="button" class="navbar-toggle"
            data-toggle="collapse" data-target=".navbar-collapse">
                <span class="sr-only">Toggle navigation</span>
                <span class="icon-bar"></span>
                <span class="icon-bar"></span>
                <span class="icon-bar"></span>
            </button>
            <a class="navbar-brand" href="/">Flasky</a>
        </div>
        <div class="navbar-collapse collapse">
            <ul class="nav navbar-nav">
                <li><a href="/">Home</a></li>
            </ul>
        </div>
    </div>
</div>
{% endblock %}
{% block content %}
<div class="container">
    {% for message in get_flashed_messages() %}
    <div class="alert alert-warning">
        <button type="button" class="close" data-dismiss="alert">&times;</button>
        {{ message }}
    </div>
    {% endfor %}
    {% block page_content %}{% endblock %}
</div>
{% endblock %}

app/templates/base.html

<!DOCTYPE html>
{% extends "base.html" %}
{% import "bootstrap/wtf.html" as wtf %}
{% block title %}Flasky{% endblock %}
{% block page_content %}
<div class="page-header">
    <h1>Hello, {% if name %}{{ name }}{% else %}Stranger{% endif %}!</h1>
    {% if not known %}
    <p>Pleased to meet you!</p>
    {% else %}
    <p>Happy to see you again!</p>
    {% endif %}
</div>
{{ wtf.quick_form(form) }}
{% endblock %}

app/templates/index.html

<!DOCTYPE html>
{% extends "base.html" %}
{% block title %}Flasky - Page Not Found{% endblock %}
{% block page_content %}
<div class="page-header">
    <h1>Not Found!</h1>
</div>
{% endblock %}

app/templates/404.html

"/app/main/templates/mail" 邮件模版

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
    User <b>{{ user.username }}</b> has joined.
</head>
<body>

</body>
</html>

app/templates/mail/new_user.html

User {{ user.username }} has joined.

app/templates/mail/new_user.txt

"/app/main/static/favicon.ico" 静态 icon 图片文件

创建需求文件

程序中必须包含一个 requirements.txt 文件，用于记录所有依赖包及其精确的版本号。如果要在另一台电脑上重新生成虚拟环境，这个文件的重要性就体现出来了，例如部署程序时使用的电脑。

(venv) E:\flasky>pip3 freeze > requirements.txt

创建数据库

(venv) E:\flasky>python manage.py shell
>>> db.create_all()
>>> exit()

生成数据库迁移文件

(venv) E:\flasky>python manage.py db init
Creating directory E:\flasky\migrations ... done
Creating directory E:\flasky\migrations\versions ... done
Generating E:\flasky\migrations\alembic.ini ... done
Generating E:\flasky\migrations\env.py ... done
Generating E:\flasky\migrations\README ... done
Generating E:\flasky\migrations\script.py.mako ... done
Please edit configuration/connection/logging settings in \'E:\\flasky\\migrations\\alembic.ini\' before proceeding.

(venv) E:\flasky>python manage.py db migrate -m "initial migration"
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.env] No changes in schema detected.

(venv) E:\flasky>python manage.py db upgrade
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.

运行测试

(venv) E:\flasky>python manage.py test
test_app_exists (test_basics.BasicsTestCase)
确保程序实例存在 ... ok
test_app_is_testing (test_basics.BasicsTestCase)
确保程序在测试中运行 ... ok

----------------------------------------------------------------------
Ran 2 tests in 2.232s

OK

启动程序

(venv) E:\flasky>python manage.py runserver
 * Serving Flask app "app" (lazy loading)
 * Environment: production
   WARNING: Do not use the development server in a production environment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 138-639-525
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

浏览器输入 http://127.0.0.1:5000

输入用户名并提交：

程序会异步发送邮件，程序控制台会打印发送日志。已收到邮件：