Flask 实战教程：用 SQLite 定义和访问 Flaskr 的数据库（db.py、schema.sql 与 init-db 命令）

本文基于 Flask 官方教程中的数据库章节（database.rst），完整讲解如何为 Flaskr 博客应用接入 SQLite 数据库：从基于请求生命周期的连接管理（get_db/close_db）、表结构的 SQL 定义，到 init-db 命令行命令的注册与执行。读完本篇，你将掌握 Python 标准库 sqlite3 与 Flask 应用上下文（g、current_app、teardown 机制）配合使用数据库的完整模式，并能在 examples/tutorial 目录下的真实示例代码中找到每一步的实现与测试佐证。

为什么选择 SQLite

Flaskr 应用使用 SQLite 数据库来存储用户（users）和帖子（posts）。Python 自带的 sqlite3 模块提供了对 SQLite 的内建支持，无需任何第三方驱动。

选择 SQLite 的主要原因是它不需要搭建独立的数据库服务器，且直接内建于 Python 中，对小型应用非常友好。但需要注意其局限性：当并发请求同时写入数据库时，各写入操作会顺序执行，从而产生等待和变慢。小型应用不会察觉这种影响，而当应用规模变大后，可能需要迁移到其他数据库（如 PostgreSQL、MySQL）。教程本身不涉及 SQL 语法细节，对 SQL 不熟悉的读者可以查阅 SQLite 官方文档中的语言说明。

连接数据库：与请求生命周期绑定的连接对象

操作 SQLite（以及大多数 Python 数据库库）的第一步是创建连接（connection）——所有查询和操作都通过连接执行，工作完成后再关闭连接。在 Web 应用中，这个连接通常与请求绑定：在处理请求的某个时刻创建，在响应发出之前关闭。

Flaskr 在 flaskr/db.py 中实现了这一模式：

# flaskr/db.py
import sqlite3
from datetime import datetime

import click
from flask import current_app, g


def get_db():
    if "db" not in g:
        g.db = sqlite3.connect(
            current_app.config["DATABASE"],
            detect_types=sqlite3.PARSE_DECLTYPES,
        )
        g.db.row_factory = sqlite3.Row

    return g.db


def close_db(e=None):
    db = g.pop("db", None)

    if db is not None:
        db.close()

逐行拆解这段核心代码的四个关键对象：

• g：Flask 的特殊对象，每个请求独享一份，用于存储请求期间可能被多个函数访问的数据。连接被保存在 g 中并复用——同一个请求里第二次调用 get_db() 时，不会创建新连接，而是直接返回已存在的 g.db。
• current_app：指向当前正在处理请求的 Flask 应用实例的特殊对象。由于 Flaskr 使用应用工厂（application factory），编写 db.py 的代码时应用实例尚不存在；但 get_db 一定是在应用已创建、正在处理请求时才被调用，因此可以用 current_app 安全地访问配置。
• sqlite3.connect(...)：建立到 DATABASE 配置项所指向文件的连接。注意这个文件此刻还不要求存在——它会在稍后初始化数据库时才被创建。
• sqlite3.Row：设置行工厂，使连接返回的行表现得像字典，从而可以按列名访问数据（如 row["username"]），而不是只靠位置索引。

close_db 则通过检查 g.db 是否被设置来判断该请求是否创建过连接；若连接存在就关闭它。稍后会把 close_db 注册到应用上，使它在每个请求结束时被调用。

close_db 的第一个参数 e 是预留的错误对象——当请求抛出异常时，teardown 函数会收到该错误（详见后文 src/flask/sansio/app.py 中 teardown_appcontext 的说明），本例中并未使用它。

DATABASE 配置项在哪里定义

连接目标来自 current_app.config["DATABASE"]。在应用工厂 flaskr/__init__.py 中，该配置被设置为实例文件夹（instance folder）下的 flaskr.sqlite：

app = Flask(__name__, instance_relative_config=True)
app.config.from_mapping(
    SECRET_KEY="dev",
    # store the database in the instance folder
    DATABASE=os.path.join(app.instance_path, "flaskr.sqlite"),
)

实例文件夹是 Flask 专门用于存放本地数据（数据库、配置等不应提交到版本控制的文件）的目录，工厂中还会用 os.makedirs(app.instance_path, exist_ok=True) 确保其存在。

创建数据表：schema.sql

在 SQLite 中，数据存储在**表（tables）和列（columns）**中，必须先创建表才能存储和读取数据。Flaskr 把用户存到 user 表、帖子存到 post 表。仓库中的真实文件 flaskr/schema.sql 内容如下：

-- Initialize the database.
-- Drop any existing data and create empty tables.

DROP TABLE IF EXISTS user;
DROP TABLE IF EXISTS post;

CREATE TABLE user (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  username TEXT UNIQUE NOT NULL,
  password TEXT NOT NULL
);

CREATE TABLE post (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  author_id INTEGER NOT NULL,
  created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  title TEXT NOT NULL,
  body TEXT NOT NULL,
  FOREIGN KEY (author_id) REFERENCES user (id)
);

表结构要点：

表	列	说明
user	id	整型主键，自增
	username	文本，UNIQUE 约束保证用户名唯一，NOT NULL
	password	文本，NOT NULL（后续教程章节会改为存散列值）
post	id	整型主键，自增
	author_id	外键，引用 user(id)，NOT NULL
	created	TIMESTAMP，默认值 CURRENT_TIMESTAMP
	title / body	文本，NOT NULL

开头的 DROP TABLE IF EXISTS 使 init-db 可以重复执行：每次运行都会先清空现有数据再重建空表。

初始化数据库：init_db、CLI 命令与时间戳转换器

接下来向 db.py 添加执行上述 SQL 的 Python 函数：

# flaskr/db.py（续）
def init_db():
    db = get_db()

    with current_app.open_resource("schema.sql") as f:
        db.executescript(f.read().decode("utf8"))


@click.command("init-db")
def init_db_command():
    """Clear existing data and create new tables."""
    init_db()
    click.echo("Initialized the database.")


sqlite3.register_converter(
    "timestamp", lambda v: datetime.fromisoformat(v.decode())
)

三个函数各司其职：

• init_db()：current_app.open_resource("schema.sql") 打开相对于 flaskr 包根目录的资源文件——使用它的好处是部署时你不必知道包的绝对安装位置，Flask 会替你定位。src/flask/app.py 中的 open_resource 实现表明它打开相对于 root_path 的文件且只支持只读模式（"r"、"rt" 或 "rb"）。拿到的数据库连接随后用 executescript 执行文件里读出的全部 SQL 命令（executescript 可以一次执行多条语句，正好对应 schema.sql 中的 DROP + CREATE 序列）。
• init_db_command()：@click.command("init-db") 定义了一个名为 init-db 的命令行命令，调用 init_db 并向用户输出成功提示。更多编写 CLI 命令的知识可参考 Flask CLI 文档。
• sqlite3.register_converter：告诉 Python 如何解释数据库中的 timestamp 值——把 ISO 格式字符串解码并转换成 datetime.datetime 对象。注意它与 get_db 中 sqlite3.connect(..., detect_types=sqlite3.PARSE_DECLTYPES) 是配套使用的：PARSE_DECLTYPES 让驱动根据列声明类型（如 TIMESTAMP）去查找已注册的转换器，缺省转换时 row["created"] 得到的就是 datetime 对象而非字符串。

注册到应用：init_app 与工厂调用

close_db 和 init_db_command 必须注册到应用实例上才会生效。但由于使用工厂函数，编写这些函数时应用实例还不可用。因此，写一个接收应用、完成注册的函数：

# flaskr/db.py（续）
def init_app(app):
    app.teardown_appcontext(close_db)
    app.cli.add_command(init_db_command)

• app.teardown_appcontext(close_db)：告诉 Flask 在返回响应后清理时调用该函数。从 src/flask/sansio/app.py 的源码文档可以看到，teardown 函数在应用上下文（app context）被弹出时调用——即请求结束、CLI 命令结束或手动 with app.app_context(): 块退出之时；且当 teardown 是因未处理异常而触发时，函数会收到该错误对象，这正是 close_db(e=None) 保留错误参数的原因。
• app.cli.add_command(init_db_command)：向应用的 flask 命令组添加一条新命令，使 flask --app flaskr init-db 可以调用它。

然后在工厂函数中导入并调用它。真实工厂代码（flaskr/__init__.py）中该调用位于创建应用与配置加载之后、注册蓝图之前：

def create_app(test_config=None):
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_mapping(...)
    # existing code omitted ...

    # register the database commands
    from . import db
    db.init_app(app)

    # apply the blueprints to the app
    from . import auth
    from . import blog
    app.register_blueprint(auth.bp)
    app.register_blueprint(blog.bp)

    app.add_url_rule("/", endpoint="index")
    return app

注意工厂函数把 from . import db 推迟到函数内部执行，避免模块导入阶段的循环依赖与提前求值。

初始化数据库文件：运行 init-db 命令

init-db 已注册到应用后，就可以像之前教程中的 flask run 一样，通过 flask 命令调用它。

提示：如果上一节的开发服务器还在运行，可以先停掉它，或者打开一个新的终端运行此命令。若使用新终端，记得先切换到项目目录并激活虚拟环境（参见 安装指南）。

$ flask --app flaskr init-db
Initialized the database.

执行成功后，项目的 instance 文件夹中会出现 flaskr.sqlite 文件——这就是前面 DATABASE 配置指向、但此前尚未创建的那个数据库文件，里面已建好空的 user 表和 post 表。

测试佐证：连接复用与命令调用的自动化验证

仓库示例自带了针对数据库模块的测试 tests/test_db.py，直接验证了本文讲解的两个核心行为：

def test_get_close_db(app):
    with app.app_context():
        db = get_db()
        assert db is get_db()          # 同一上下文内复用同一连接

    with pytest.raises(sqlite3.ProgrammingError) as e:
        db.execute("SELECT 1")        # 上下文退出后 teardown 已关闭连接

    assert "closed" in str(e.value)


def test_init_db_command(runner, monkeypatch):
    ...
    result = runner.invoke(args=["init-db"])
    assert "Initialized" in result.output

第一个测试印证了 get_db 的复用语义（assert db is get_db()）以及 teardown_appcontext(close_db) 的生效时机：with app.app_context(): 块退出时 close_db 被调用，之后再执行 SQL 会抛出 "closed" 错误。第二个测试则通过 Flask 提供的 test_cli_runner（见 tests/conftest.py 中的 runner fixture）模拟 flask init-db 的调用，确认命令输出 Initialized 且 init_db 确实被执行。该 fixture 还会为每个测试创建临时数据库文件（tempfile.mkstemp()）并注入 DATABASE 配置，使测试与开发用的 instance/flaskr.sqlite 完全隔离。

小结

本章节完整构建了 Flaskr 的数据库层：

1. 连接管理：get_db() 以 g 为载体实现“每请求一连接、多次调用复用”，close_db 通过 teardown_appcontext 在请求（或 app context）结束时自动关闭，保证连接不泄漏；
2. 表结构：schema.sql 用 DROP TABLE IF EXISTS + CREATE TABLE 定义可重复初始化的 user/post 两表，外键把帖子关联到作者；
3. 资源加载：open_resource 使 schema.sql 相对包目录定位，部署位置无关；
4. CLI 入口：@click.command("init-db") + app.cli.add_command 让 flask --app flaskr init-db 一键初始化，并在 instance 文件夹生成 flaskr.sqlite；
5. 类型转换：sqlite3.register_converter 配合 PARSE_DECLTYPES，让 TIMESTAMP 列直接以 datetime 对象返回。

完成本节后，数据库基础设施已就绪，教程的下一步是编写访问这些数据的视图函数，参见 views.rst。