Flask 数据库集成

大多数 Web 应用都需要持久化存储数据。

本章使用 Python 内置的 SQLite 数据库，展示如何在 Flask 中集成数据库，无需安装额外依赖。

SQLite 快速入门

SQLite 是一个轻量级数据库，数据存储在一个文件中，Python 标准库自带支持。

它是学习和原型开发的最佳选择——无需安装数据库服务器，无需配置。

实例

# 一个简单的 SQLite 示例（不使用 Flask）
import sqlite3

# 连接数据库（如果 runoob.db 不存在则自动创建）
conn = sqlite3.connect("runoob.db")
# row_factory 让查询结果像字典一样可用字段名访问
conn.row_factory = sqlite3.Row

# 创建表
conn.execute("CREATE TABLE IF NOT EXISTS user (id INTEGER PRIMARY KEY, username TEXT UNIQUE, password TEXT)")
# 插入数据
conn.execute("INSERT INTO user (username, password) VALUES (?, ?)", ("runoob", "pass123"))
conn.commit()

# 查询数据
rows = conn.execute("SELECT * FROM user").fetchall()
for row in rows:
print(dict(row)) # 输出: {'id': 1, 'username': 'runoob', 'password': 'pass123'}

conn.close()

在 Flask 中使用 g 对象管理数据库连接

g 是一个特殊的对象，它的生命周期是单个请求。

将数据库连接存入 g，可以确保同一请求内的多次数据库操作共享同一连接，请求结束后自动清理。

实例

# 文件路径：db.py（数据库管理模块）
import sqlite3
from flask import g, current_app

def get_db():
"""获取数据库连接，同一请求内复用"""
if "db" not in g:
# 从配置中读取数据库路径
db_path = current_app.config.get("DATABASE", "runoob.db")
g.db = sqlite3.connect(db_path)
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()

def init_db():
"""初始化数据库表结构"""
db = get_db()
db.execute("""
CREATE TABLE IF NOT EXISTS user (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS post (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
body TEXT NOT NULL,
author_id INTEGER NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (author_id) REFERENCES user (id)
)
""")
db.commit()

def init_app(app):
"""在 Flask 应用中注册数据库相关功能"""
# 请求结束时自动关闭数据库连接
app.teardown_appcontext(close_db)
# 注册 CLI 命令：flask init-db
app.cli.command("init-db")(init_db)

工厂模式整合

使用工厂模式将数据库与其他模块整合：

实例

# 文件路径：app.py
from flask import Flask

def create_app():
app = Flask(__name__)
app.secret_key = "dev-secret-key"

# 配置数据库路径（存储在 instance 文件夹中）
import os
app.config["DATABASE"] = os.path.join(app.instance_path, "runoob.db")

# 确保 instance 文件夹存在
os.makedirs(app.instance_path, exist_ok=True)

# 初始化数据库
from db import init_app as init_db
init_db(app)

# 注册蓝图
from auth import bp as auth_bp
from blog import bp as blog_bp
app.register_blueprint(auth_bp, url_prefix="/auth")
app.register_blueprint(blog_bp)

return app

写一个完整的博客 API

结合数据库，实现文章的增删改查（CRUD）操作：

实例

# 文件路径：blog.py
from flask import Blueprint, request, jsonify, abort, g
from db import get_db

bp = Blueprint("blog", __name__)

@bp.get("/api/posts")
def list_posts():
"""获取所有文章列表"""
db = get_db()
posts = db.execute(
"SELECT p.id, p.title, p.body, p.created_at, u.username"
" FROM post p JOIN user u ON p.author_id = u.id"
" ORDER BY p.created_at DESC"
).fetchall()
# 将 Row 对象列表转为 dict 列表
return jsonify([dict(post) for post in posts])

@bp.get("/api/posts/<int:post_id>")
def get_post(post_id):
"""获取单篇文章详情"""
db = get_db()
post = db.execute(
"SELECT id, title, body, created_at FROM post WHERE id = ?",
(post_id,)
).fetchone()

if post is None:
abort(404, description=f"文章 {post_id} 不存在")

return jsonify(dict(post))

@bp.post("/api/posts")
def create_post():
"""创建新文章"""
data = request.json
title = data.get("title", "").strip()
body = data.get("body", "").strip()
author_id = data.get("author_id", 1) # 默认作者 ID

if not title:
return jsonify({"error": "标题不能为空"}), 400

db = get_db()
cursor = db.execute(
"INSERT INTO post (title, body, author_id) VALUES (?, ?, ?)",
(title, body, author_id)
)
db.commit()

return jsonify({"id": cursor.lastrowid, "title": title, "message": "文章创建成功"}), 201

@bp.put("/api/posts/<int:post_id>")
def update_post(post_id):
"""更新文章"""
data = request.json
title = data.get("title", "").strip()
body = data.get("body", "").strip()

if not title:
return jsonify({"error": "标题不能为空"}), 400

db = get_db()
db.execute(
"UPDATE post SET title = ?, body = ? WHERE id = ?",
(title, body, post_id)
)
db.commit()

return jsonify({"id": post_id, "message": "文章更新成功"})

@bp.delete("/api/posts/<int:post_id>")
def delete_post(post_id):
"""删除文章"""
db = get_db()
db.execute("DELETE FROM post WHERE id = ?", (post_id,))
db.commit()
return jsonify({"message": "文章已删除"})

测试 API：

# 获取文章列表
$ curl http://127.0.0.1:5000/api/posts

# 创建文章
$ curl -X POST http://127.0.0.1:5000/api/posts \
  -H "Content-Type: application/json" \
  -d '{"title":"RUNOOB 教程","body":"Flask 入门教程内容","author_id":1}'

# 更新文章
$ curl -X PUT http://127.0.0.1:5000/api/posts/1 \
  -H "Content-Type: application/json" \
  -d '{"title":"RUNOOB 教程（修订版）","body":"更新后的内容"}'

# 删除文章
$ curl -X DELETE http://127.0.0.1:5000/api/posts/1

最佳实践：API 返回 JSON 数据时，使用正确的 HTTP 状态码——创建用 201，成功用 200，客户端错误用 400，资源不存在用 404。这能让前端代码更容易判断请求结果。

数据库初始化命令

通过 Flask CLI 可以创建自定义命令来初始化数据库：

# 初始化数据库（创建表结构）
(.venv) $ flask init-db

# 验证表已创建
(.venv) $ sqlite3 instance/runoob.db ".tables"
post  user

SQL 注入防护

使用参数化查询（? 占位符）是防止 SQL 注入的关键：

实例

# 正确做法：使用 ? 占位符（参数化查询）
# SQLite 驱动会自动转义参数，防止恶意输入被当作 SQL 执行
db.execute("SELECT * FROM user WHERE username = ?", (username,))

# 错误做法：用 f-string 拼接 SQL（极其危险！）
# db.execute(f"SELECT * FROM user WHERE username = '{username}'")
# 恶意输入 '; DROP TABLE user; -- 会删除整个表！

使用 Flask-SQLAlchemy（进阶选读）

对于大型项目，手动写 SQL 会逐渐变得繁琐。

Flask-SQLAlchemy 是一个广受欢迎的扩展，提供了 ORM（对象关系映射）能力：

实例

# 安装：pip install flask-sqlalchemy
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///runoob.db"
db = SQLAlchemy(app)

# 定义数据模型（表结构用 Python 类表示）
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)

def __repr__(self):
return f"<User {self.username}>"

# 创建表
with app.app_context():
db.create_all()

# 创建用户
user = User(username="runoob", email="test@runoob.com")
db.session.add(user)
db.session.commit()

# 查询用户
users = User.query.all()
runoob_user = User.query.filter_by(username="runoob").first()

本章使用原生 SQLite 是为了不引入额外依赖，适合学习和小型项目。对于实际生产项目，建议使用 SQLAlchemy 或类似 ORM，能显著减少重复代码并提升安全性。

返回顶部

菜鸟教程