注 本篇博客主要参考图灵程序设计丛书《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
,以便在不需要跟踪对象变化时降低内存消耗
1 2 3 4 5 6 7 8 9 10 11 12 13 14 import osfrom flask_sqlalchemy import SQLAlchemybasedir = 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创建的数据库实例为模型提供了一个基类及一系列辅助类和辅助函数,可用于定义模型的结构
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 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
为列定义默认值
关系 关系型数据库使用关系将不同表中的行联系起来,如
1 2 3 4 5 6 7 8 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()
方法提交会话
这段代码都可以写在if __name__ == '__main__':
中:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 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
接着可以查询数据库看看结果
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 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
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 @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 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 {% 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 Migratemigrate = Migrate(app, db)
为了开发数据库迁移相关的命令,Flask-Migrate 添加了 flask db
命令和几个子命令 在新项目中可以使用init
子命令添加数据库迁移支持
1 2 3 4 5 6 7 8 (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 子命令用于自动创建迁移脚本
1 2 3 4 (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 应用的基本结构示例如下
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 ├─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
:应用的配置1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 import osbasedir = 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
:应用包的构造文件1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 from flask import Flaskfrom flask_bootstrap import Bootstrapfrom flask_mail import Mailfrom flask_moment import Momentfrom flask_sqlalchemy import SQLAlchemyfrom config import configbootstrap = 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
:创建主蓝本1 2 3 4 5 from flask import Blueprintmain = 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
:注册主蓝本1 2 3 4 5 6 def create_app (config_name) : from .main import main as main_blueprint app.register_blueprint(main_blueprint) return app
app/main/errors.py
:主蓝本中的错误处理程序1 2 3 4 5 6 7 8 9 10 11 12 from flask import render_templatefrom . 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
:主蓝本中定义的应用路由1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from flask import render_template, session, redirect, url_for, current_appfrom .. import dbfrom ..models import Userfrom ..email import send_emailfrom . import mainfrom .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
:主脚本1 2 3 4 5 6 7 8 9 10 11 12 import osfrom flask_migrate import Migratefrom app import create_app, dbfrom app.models import User, Roleapp = 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
:单元测试1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 import unittestfrom flask import current_appfrom app import create_app, dbclass 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
:启动单元测试的命令1 2 3 4 5 6 @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