给 Flask 应用增加数据库，真正需要处理的不只是一次 INSERT。配置应该从环境变量读取，数据库连接需要跟随应用上下文关闭，路由要按 Blueprint 组织，表单提交还应经过验证，并在写入后重定向。下面用一个最小博客项目把这些环节串起来。
数据库之外，还有四个边界要处理
一个能长期维护的 Flask 数据写入流程，通常包含以下职责：
- 环境变量：保存密钥、数据库路径等部署相关配置，避免写死在仓库里。
- Blueprint：把文章相关路由放在独立模块边界中，避免所有接口堆进应用工厂。
- Schema：显式定义字段、约束和默认值，让数据库承担一部分数据完整性检查。
- 请求处理：验证表单、执行参数化 SQL，并采用 Post/Redirect/Get 模式防止刷新时重复提交。
SQLite 很适合本地工具、教程项目和低写入并发的服务。它不需要单独启动数据库进程，但这并不意味着可以忽略连接管理、迁移和备份。
一个可以直接运行的最小项目
创建一个新目录，并准备以下两个文件：
flask-posts/
├── app.py
└── schema.sql
将下面内容保存为 schema.sql：
CREATE TABLE IF NOT EXISTS post (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL CHECK (length(trim(title)) > 0),
body TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
再将下面代码保存为 app.py。为便于复制运行，Blueprint、数据库函数和应用工厂暂时放在同一个文件中；项目变大后可以分别拆到 blog.py、db.py 和包初始化文件中。
import os
import sqlite3
from pathlib import Path
import click
from flask import (
Blueprint,
Flask,
current_app,
flash,
g,
redirect,
render_template_string,
request,
url_for,
)
blog = Blueprint('blog', __name__)
def get_db():
if 'db' not in g:
g.db = sqlite3.connect(current_app.config['DATABASE'])
g.db.row_factory = sqlite3.Row
return g.db
def close_db(error=None):
db = g.pop('db', None)
if db is not None:
db.close()
@click.command('init-db')
def init_db_command():
schema_path = Path(current_app.root_path) / 'schema.sql'
get_db().executescript(schema_path.read_text(encoding='utf-8'))
click.echo('Database initialized.')
@blog.route('/')
def index():
posts = get_db().execute(
'SELECT id, title, body, created_at '
'FROM post ORDER BY id DESC'
).fetchall()
return render_template_string(
'''
<!doctype html>
<meta charset='utf-8'>
<title>Posts</title>
<h1>Posts</h1>
<p><a href='{{ url_for("blog.create") }}'>Create a post</a></p>
{% for message in get_flashed_messages() %}
<p><strong>{{ message }}</strong></p>
{% endfor %}
{% for post in posts %}
<article>
<h2>{{ post['title'] }}</h2>
<p>{{ post['body'] }}</p>
<small>{{ post['created_at'] }}</small>
</article>
{% else %}
<p>No posts yet.</p>
{% endfor %}
''',
posts=posts,
)
@blog.route('/posts/new', methods=['GET', 'POST'])
def create():
if request.method == 'POST':
title = request.form.get('title', '').strip()
body = request.form.get('body', '').strip()
if not title:
flash('Title is required.')
else:
db = get_db()
db.execute(
'INSERT INTO post (title, body) VALUES (?, ?)',
(title, body),
)
db.commit()
flash('Post created.')
return redirect(url_for('blog.index'))
return render_template_string(
'''
<!doctype html>
<meta charset='utf-8'>
<title>New post</title>
<h1>New post</h1>
{% for message in get_flashed_messages() %}
<p><strong>{{ message }}</strong></p>
{% endfor %}
<form method='post'>
<p><label>Title <input name='title' required></label></p>
<p><label>Body<br><textarea name='body' rows='8'></textarea></label></p>
<button type='submit'>Save</button>
</form>
<p><a href='{{ url_for("blog.index") }}'>Back</a></p>
'''
)
def create_app():
app = Flask(__name__, instance_relative_config=True)
app.config.from_mapping(
SECRET_KEY=os.getenv('FLASK_SECRET_KEY', 'dev-only-change-me'),
DATABASE=os.getenv(
'DATABASE_PATH',
str(Path(app.instance_path) / 'posts.sqlite'),
),
)
Path(app.config['DATABASE']).parent.mkdir(parents=True, exist_ok=True)
app.teardown_appcontext(close_db)
app.cli.add_command(init_db_command)
app.register_blueprint(blog)
return app
app = create_app()
安装 Flask、设置环境变量并启动服务：
python -m venv .venv
source .venv/bin/activate
python -m pip install Flask
export FLASK_SECRET_KEY='replace-with-a-random-secret'
export DATABASE_PATH="$PWD/instance/posts.sqlite"
flask --app app init-db
flask --app app run --debug
Windows PowerShell 可以把两条 export 改成：
$env:FLASK_SECRET_KEY = 'replace-with-a-random-secret'
$env:DATABASE_PATH = "$PWD\instance\posts.sqlite"
启动后访问 http://127.0.0.1:5000/posts/new，提交的文章会写入 SQLite，并在首页显示。
为什么这些细节值得保留
get_db() 把连接存进 Flask 的 g 对象，因此同一个请求可以复用连接；teardown_appcontext() 则确保请求结束后关闭它。直接在模块导入时创建全局 SQLite 连接，会给测试、线程和应用工厂带来额外问题。
SQL 使用 ? 占位符，而不是字符串拼接：
db.execute(
'INSERT INTO post (title, body) VALUES (?, ?)',
(title, body),
)
这样数据库驱动会把输入当作参数处理，避免标题中的引号破坏 SQL，也降低 SQL 注入风险。不要改成 f-string 或 % 格式化 SQL。
写入成功后返回重定向，而不是直接渲染成功页面。这是 Post/Redirect/Get 模式：浏览器刷新时重新请求的是首页，不会再次提交同一份表单。
Blueprint 的价值则会在功能增长后显现。文章、账户和管理后台可以使用不同的 Blueprint，并分别配置 URL 前缀、模板和权限检查。
上线前需要补齐什么
这个示例刻意保持最小，因此不应原样承担公开生产流量。部署前至少检查：
- 用随机且稳定的
FLASK_SECRET_KEY，不要使用示例默认值。 - 为表单增加 CSRF 防护；可以采用 Flask-WTF，或实现等价的令牌校验。
- 增加身份认证和授权，不能让所有访问者都能创建文章。
- 限制标题、正文和请求体大小，并处理数据库异常。
- 不要只靠反复执行
CREATE TABLE IF NOT EXISTS演进结构；表结构变化后应使用迁移工具。 - 定期备份数据库文件，并验证恢复流程。
- 评估并发写入量。SQLite 会序列化部分写操作，写入密集型、多实例服务通常更适合 PostgreSQL 等独立数据库。
判断是否应该继续使用 SQLite，可以看三个问题：是否只有单机或共享磁盘实例、写入是否较少、是否能接受文件级备份方式。如果答案逐渐变成否，保留当前 Blueprint 和配置边界，再替换数据访问层，会比重写整个 Flask 应用容易得多。