注

本篇博客主要参考图灵程序设计丛书《Flask Web开发》（第二版）
作者：【美】Miguel Grinberg

数据库

大多数数据库引擎都有对应的 Python 包，Flask 并不限制使用的数据包类型，可以根据自己的喜好选择
如 MySQL, Redis, SQLite, MongoDB 等等

还有一些数据库抽象层代码包供选择，如SQLAlchemy和MongoEngine

选择数据库框架时不一定非得选择已经集成了 Flask 的框架，但选择了这样的框架可以节省编写代码的时间，这里以 Flask-SQLAlchemy 为例

SQLAlchemy 提供了高层 ORM，也提供了使用数据库原生 SQL 的低层功能

安装 Flask-SQLAlchemy

1	(venv)>pip install flask-sqlalchemy

在 Flask-SQLAlchemy 中，数据库使用 URL 指定，列举两种数据库引擎

Flask-SQLAlchemy 数据库 URL

数据库引擎	URL
MySQL	mysql://username:password@hostname/databse
SQLite(Windows)	sqlite:///c:absolute/path/to/database

URL 中的database时磁盘中的文件名

配置数据库

首先在终端创建数据库flask_db

1 2	mysql>create database flask_db charset=utf8; ...

应用使用的数据库URL必须保存到Flask配置对象的 SQLALCHEMY_DATABASE_URI键中
Flask-SQLAlchemy文档建议把SQLALCHEMY_TRACK_MODIFICATIONS键设置为 False，以便在不需要跟踪对象变化时降低内存消耗

import os
from flask_sqlalchemy import SQLAlchemy

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

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:password@127.0.0.1/flask_db'
'''
app.config['SQLALCHEMY_DATABASE_URI'] =\
    'sqlite:///' + os.path.join(basedir, 'data.sqlite') # sqlite:///F:\TEMP\flasky\app\data.sqlite
'''
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

db是SQLAlchemy对象的实例，通过它可以获得Flask-SQLAlchemy提供的所有功能

定义模型

Flask-SQLAlchemy创建的数据库实例为模型提供了一个基类及一系列辅助类和辅助函数，可用于定义模型的结构

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)

    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)

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

类变量__tablename__定义在数据库使用的表名
定义的__repr()__方法，返回一个具有可读性的字符串模型，供调试和测试使用

db.Column类构造函数的第一个参数是数据库列和模型属性的类型

几种常用的 SQLAlchemy 列类型

类型名	Python类型	说明
Integer	int	普通整数，通常为32位
BigInteger	int或long	不限制精度的整数
Float	float	浮点数
String	str	变长字符串
Text	str	优化了的变长字符串
Unicode	unicode	变长Unicode字符串
UnicodeText	unicode	优化的Unicode字符串
Boolean	bool	布尔值
DataTime	datatime.datatime	日期和时间
Interval	datatime.timedelta	时间间隔

db.Column的其余参数指定属性的配置选项，下表列出了一些可用选项

选项名	说明
primary_key	如果设为True，列为表的主键
unique	如果设为True，列不允许出现重复的值
index	如果设为True，为列创建索引，提升查询效率
nullable	如果设为True，列允许使用空值，如果设为False，列不允许使用空值
default	为列定义默认值

关系

关系型数据库使用关系将不同表中的行联系起来，如

class Role(db.Model):
    # ...
    users = db.relationship('User', backref='role')


class User(db.Model):
    # ...
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

数据库操作

创建表

db.create_all()函数将寻找所有db.Model的子类，然后在数据库中创建对应的表

1
2
3

(venv)>flask shell
>>>from hello import db
>>>db.create_all()

若数据库表已经存在，那么db.create_all()不会重新创建或者更新表
更新现有数据库的蛮力方式是先删除旧表再重新创建

1 2	db.drop_all() db.create_all()

但是其有个副作用，销毁了数据库中所有的数据

插入行

下面的代码创建了一些角色和用户

1
2
3

admin_role = Role(name='Admin')
user_role = Role(name='User')
user_nopech = User(username='Nopech')

新建对象时没有明确设定id属性，因为多数数据库中主键由数据库自身管理

对数据库的改动通过数据库会话管理，在 Flask-SQLAlchemy 中，会话由 db.session表示

现在将对象添加到会话中

1
2
3

db.session.add(admin_role)
db.session.add(user_role)
db.session.add(user_nopech)

上述代码可以简写，把对象放进列表

1	db.session.add([admin_role, user_role, user_nopech])

为了把对象写入数据库，还要使用commit()方法提交会话

1	db.session.commit()

这段代码都可以写在if __name__ == '__main__':中：

if __name__ == '__main__':
    db.drop_all()
    db.create_all()

    admin_role = Role(name='Admin')
    user_role = Role(name='User')

    db.session.add(admin_role)
    db.session.add(user_role)
    db.session.commit()

    user_nopech = User(username='Nopech', role_id=admin_role.id)
    db.session.add(user_nopech)
    db.session.commit()

注意，admin_role和user_role需要先提交到数据库
因为user_nopech对象的role_id参数引用了admin_role的id
未提交之前admin_role.id = NULL

接着可以查询数据库看看结果

mysql> use flask_db;
Database changed
mysql> select * from roles;
+----+-------+
| id | name  |
+----+-------+
|  1 | Admin |
|  2 | User  |
+----+-------+
2 rows in set (0.00 sec)

mysql> select * from users;
+----+----------+---------+
| id | username | role_id |
+----+----------+---------+
|  1 | Nopech   |       1 |
+----+----------+---------+
1 row in set (0.00 sec)

修改行

1
2
3

>>> admin_role.name = 'Administrator'
>>> db.session.add(admin_role)
>>> db.session.commit()

删除行

1 2	>>> db.session.delete(mode_role) >>> db.session.commit()

查询行

Flask-SQLAlchemy 为每个模型类都提供了query对象，最基本的模型查询是使用all()方法取回对应表中的所有记录

1 2	>>> print(Role.query.all()) [<Role 'Admin'>, <Role 'User'>]

1 2	>>> User.query.filter_by(role_id=admin_role.id).all() [<User 'Nopech'>]

列举几个常用的 SQLAlchemy 查询过滤器

过滤器	说明
filter()	添加过滤器到原查询上，返回一个新查询
filter_by()	添加等值过滤器到原查询上，返回一个新查询
limit()	使用指定的值限制原查询返回的结果数量，返回一个新查询
offset()	便宜原查询返回的结果，返回一个新查询
order_by()	根据指定条件对原查询结果进行排序，返回一个新查询
group_by()	根据指定条件对原查询结果进行分组，返回一个新查询

常用的 SQLAlchemy 查询执行方法

方法	说明
all()	以列表形式返回查询的所有结果
first()	返回查询的第一个结果，如果没有结果，返回 NONE
first_or_404()	返回查询的第一个结果，如果没有结果，则终止请求，返回404错误响应
get()	返回主键对应的行，如果没有行，则返回 NONE
get_or_404	返回主键对应的行，如果没有行，则终止请求，返回404响应
count()	返回查询结果的数量

在视图函数中操作数据库

直接附上代码

`hello.py`

@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)
            db.session.commit()
            session['known'] = False
        else:
            session['known'] = True
        session['name'] = form.name.data
        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())

templates/index.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>
<p>The local date and time is {{ moment(current_time).format('LLL') }}.</p>
<p>That was {{ moment(current_time).fromNow(refresh=True) }}.</p>
{{ wtf.quick_form(form) }}
{% endblock %}

简单测试一下，提交两次表单，结果如下

使用Flask-Migrate实现数据库迁移

之前说过，仅当数据表不存在时，Flask-SQLAlchemy 才会根据模型创建，更新表就采用了暴力法，删除旧表在创建新表，但这样会丢失数据库的全部数据

更好的方法是使用数据库迁移框架

SQLAlchemy 的开发人员编写了一个迁移框架–Alembic
Flask 应用还可以使用 Flask-Migrate 扩展
这个扩展是对 Alembic 的轻量级包装，并于 flask 命令做了集成

创建迁移仓库

首先要在虚拟环境中安装 Flask-Migrate

1	(venv)> pip install flask-migrate

接着初始化 Flask-Migrate

1
2
3

from flask_migrate import Migrate
# ...
migrate = Migrate(app, db)

为了开发数据库迁移相关的命令，Flask-Migrate 添加了 flask db 命令和几个子命令
在新项目中可以使用init子命令添加数据库迁移支持

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

接着可以在 F 盘的 flask 目录文件下看到新建了一个 migrations 文件夹，所有的迁移脚本都存放在此

数据库迁移仓库中的文件要和应用的其他文件一起纳入版本控制

创建迁移脚本

使用 Flask-Migrate 管理数据库模式变化的步骤如下

对模型做必要的修改
执行 flask db migrate 命令，自动创建一个迁移脚本
检查自动生成的脚本，根据对模型的实际改动进行调整
把迁移脚本纳入版本控制
执行 flask db upgrate 命令，把应用应用迁移道数据库中

flask db migrate 子命令用于自动创建迁移脚本

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

可以看出没有变化：No changes in schema detected.

更新数据库

检查并修正好迁移脚本之后，执行 flask db upgrate 命令，把迁移应用到数据库中

1
2
3

(venv) F:\TEMP\flasky>flask db upgrade
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.

对第一个迁移来说，其作用与调用 db.create_all() 方法一样，但在后续的迁移中，flask db upgrade 命令能把改动应用道数据库中，且不影响其中保存的数据

大型应用的结构

不同于多数其他 Web 框架，Flask 并不强制要求大型项目使用特定的组织方式，应用结构的组织方式完全由开发者决定

项目结构

Flask 应用的基本结构示例如下

├─app
|  ├─templates
|  ├─static
|  ├─main
|  |  ├─errors.py
|  |  ├─forms.py
|  |  ├─views.py
|  |  └__init__.py
|  ├─email.py
|  ├─models.py
|  ├─__init__.py
├─migrations
├─tests
|   ├─test*.py
|   └__init__.py
|-venv
├─config.py
├─flasky.py
├─requirements.txt

flask 应用一般保存在名为 app 的包中
单元测试在 tests 包中编写
requirements.txt 列出了所有依赖包，便于在其他计算机中重新生成相同的虚拟环境
config.py 存储配置
flask.py 定义 Flask 应用实例，同时还有一些辅助管理应用的任务

配置选项

应用经常需要设定多个配置，开发、测试和生产环境要使用不同的数据库，这样才不会彼此影响

除了hello.py中类似字典的app.config对象外，还可以使用具有层次结构的配置类

`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'
    MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.googlemail.com')
    MAIL_PORT = int(os.environ.get('MAIL_PORT', '587'))
    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in \
        ['true', 'on', '1']
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
    FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]'
    FLASKY_MAIL_SENDER = 'Flasky Admin <flasky@example.com>'
    FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN')
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    @staticmethod
    def init_app(app):
        pass


class DevelopmentConfig(Config):
    DEBUG = True
    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://'


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
}

备注：最好不要把密码或其他机密信息写在纳入版本控制的配置文件中

为了再给应用提供一种定制配置的方式，Config类及其子类可以定义init_app()类方法，其参数为应用实例

应用包

应用包用于存放应用的所有代码、模板和静态文件，一般把包称为app，若有需求，也可以使用一个应用专属名称

使用应用工厂函数

在单个文件中开发应用很方便，但却有个很大的缺点：应用在全局作用域中创建，无法动态修改配置。运行脚本时，应用实例已经创建，再修改配置为时已晚。这点对单元测试特别重要，因为有时为了提高测试覆盖度，必须在不同的配置下运行应用

解决方法：延迟创建应用实例，把创建过程移到可显示调用的工厂函数中
这样不仅可以给脚本留出配置应用的时间，还能够创建多个应用实例，为测试提供便利

应用的工厂函数在 app 包的构造文件中定义

`app/init.py`：应用包的构造文件

from flask import Flask
from flask_bootstrap import Bootstrap
from flask_mail import Mail
from flask_moment import Moment
from flask_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

在蓝本中实现应用功能

转换成应用工厂函数的操作让定义路由变复杂了。
应用在运行时创建，只有调用create_app()之后才能使用app.route装饰器

Flask 使用蓝本（blueprint）提供了解决办法

蓝本可以定义路由和错误处理程序，不过定义的路由和错误处理程序处于休眠状态，直到蓝本注册到应用上后，它们才能真正成为应用的一部分

蓝本可以在单个文件中定义，也可以使用更结构化的方式在包中的多个模块创建

为了获得最大的灵活性，我们将在应用包中创建一个子包，用于保存应用的第一个蓝本

`app/main/init.py`：创建主蓝本

from flask import Blueprint

main = Blueprint('main', __name__)

from . import views, errors

Blueprint类的构造函数有两个必须指定的参数：
蓝本的名称
蓝本所在的包或模块

应用的路由保存在包里的app/mian/views.py
错误处理程序保存在app/main/errors.py

这些模块在app/main/__init__.py脚本的末尾导入，是为了避免循环导入依赖
因为在app/mian/views.py和app/main/errors.py中还要导入main蓝本

`app/init.py`：注册主蓝本

def create_app(config_name):
    # ...
    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)

    return app

`app/main/errors.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

`app/main/views.py`：主蓝本中定义的应用路由

from flask import render_template, session, redirect, url_for, current_app
from .. import db
from ..models import User
from ..email import send_email
from . import main
from .forms import NameForm


@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))

路由装饰器由蓝本提供，因此使用的是main.route

应用脚本

`flask.py`：主脚本

import os
from flask_migrate import Migrate
from app import create_app, db
from app.models import User, Role

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, Role=Role)

需求文件

应用中最好有一个requirements.txt文件，用于记录所有依赖包及精确的版本号
这个文件可由pip自动生成

1	(venv) F:\TEMP\flasky>pip freeze >requirements.txt

执行完毕后就可以在目录下看到该文件，打开后就可以看到类似如下的信息

如果想创建这个虚拟环境的完整副本，就可以在新创建的虚拟环境下执行如下命令：

1	pip install -r requirements.txt

单元测试

编写两个简单的测试

`test/test_basics.py`：单元测试

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'])

为了运行单元测试，可以在flask.py中添加一个自定义命令

`flask.py`：启动单元测试的命令

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

单元测试可以用如下命令运行：

1	(venv) F:\flasky>flask test

注

数据库