给后端的数据库快速入门
2954 字约 10 分钟
2026-08-14
可以。这一块非常重要，因为到数据库之后，你的 Flask 应用才开始真正拥有“持久状态”。
先建立最核心的心智模型：
Frontend
│
│ HTTP
▼
Flask Route
│
▼
Service / Python Logic
│
▼
SQLAlchemy ORM
│
▼
Database例如用户调用：
POST /api/usersFlask 不再只是计算一个结果然后返回，而是：
收到请求
↓
创建 User
↓
写入数据库
↓
以后还能查询出来我们从最底层开始。
1. 数据库到底是什么？
假设你要保存用户：
id | name | email
---+-------+----------------
1 | Alice | alice@test.com
2 | Bob | bob@test.com这就是一张 table（表）。
其中：
User table
│
├── column: id
├── column: name
└── column: email每一条：
1 | Alice | alice@test.com叫一行：
row / record
所以最基础的关系数据库模型就是：
Database
│
├── users
│ ├── row
│ ├── row
│ └── row
│
├── orders
│
└── products2. Flask 本身不提供数据库
Flask 是 Web framework。
它负责：
HTTP request
↓
Python function
↓
HTTP response至于数据存在哪里，Flask 并不规定。
你可以使用：
SQLite
PostgreSQL
MySQL
...Flask 官方也把 Flask-SQLAlchemy 作为 SQLAlchemy 与 Flask 集成的一种常见扩展方式。(Flask 文档)
我们入门先选择：
SQLite
因为 SQLite 不需要运行独立数据库服务器，它是一个嵌入式、serverless 的 SQL 数据库，整个数据库可以直接存放在普通磁盘文件中，非常适合学习和小型项目。(SQLite)
3. SQLite 长什么样？
例如：
project/
├── app.py
└── instance/
└── app.db这个：
app.db就是数据库。
它不是 JSON：
{
"users": [...]
}而是真正的关系数据库文件。
里面可以有：
app.db
│
├── users
├── orders
└── products4. 数据库真正执行的是 SQL
例如创建用户表：
CREATE TABLE user (
id INTEGER PRIMARY KEY,
name VARCHAR(100),
email VARCHAR(200)
);插入：
INSERT INTO user (name, email)
VALUES ('Alice', 'alice@test.com');查询：
SELECT *
FROM user;更新：
UPDATE user
SET name = 'Alice Smith'
WHERE id = 1;删除：
DELETE FROM user
WHERE id = 1;这四种操作极其重要：
Create
Read
Update
Delete合起来简称：
CRUD
以后你会天天看到这个词。
5. CRUD 和 REST API 恰好可以对应起来
比如 User：
数据库操作 HTTP
Create POST
Read GET
Update PATCH / PUT
Delete DELETE于是：
POST /api/users
GET /api/users/1
PATCH /api/users/1
DELETE /api/users/1后面实际上对应：
INSERT
SELECT
UPDATE
DELETE你可以把 REST API 和数据库联系起来：
HTTP Database
POST /users → INSERT
GET /users/1 → SELECT
PATCH /users/1 → UPDATE
DELETE /users/1 → DELETE这就是大量普通 Web 后端的基本工作模式。
6. SQLAlchemy 是什么？
现在有一个问题。
难道 Flask 里每次都写：
sql = """
SELECT *
FROM user
WHERE id = ?
"""当然可以。
但 Python 项目通常会使用：
ORM：Object Relational Mapping
SQLAlchemy 提供 ORM 能力；官方 SQLAlchemy 2.x 文档的核心模式就是把数据库表映射成 Python class，再通过 Session 持久化和查询 Python 对象。(SQLAlchemy 文档)
比如数据库：
user
-------------------
id
name
email映射成：
class User:
id
name
email于是数据库里的一行：
1 | Alice | alice@test.com在 Python 看起来像：
user.id
user.name
user.email即：
database row
⇅
Python object这就是 ORM 最核心的思想。
7. Flask-SQLAlchemy 又是什么？
关系是：
SQLite / PostgreSQL / MySQL
↑
SQLAlchemy
↑
Flask-SQLAlchemy
↑
FlaskSQLAlchemy 是真正负责数据库抽象和 ORM 的库。
Flask-SQLAlchemy 是：
帮你把 SQLAlchemy 更方便地集成进 Flask。
它会帮 Flask 项目配置 engine、session 等常见对象，但并没有改变 SQLAlchemy 本身的工作方式。(Flask-SQLAlchemy)
安装：
pip install Flask-SQLAlchemy官方当前 Flask-SQLAlchemy 3.1.x 文档假定使用 SQLAlchemy 2.x。(Flask-SQLAlchemy)
8. 第一个 Flask 数据库项目
先写最简单版本。
project/
└── app.pyapp.py：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
db = SQLAlchemy(app)这里：
sqlite:///app.db表示：
使用 SQLite 数据库。
而：
db = SQLAlchemy(app)把 SQLAlchemy 与 Flask application 绑定起来。
9. 定义第一张表
接下来定义 User：
class User(db.Model):
id = db.Column(
db.Integer,
primary_key=True
)
name = db.Column(
db.String(100),
nullable=False
)
email = db.Column(
db.String(200),
unique=True,
nullable=False
)你可以把它直接翻译成：
User
│
├── id
│ Integer
│ Primary Key
│
├── name
│ String
│ NOT NULL
│
└── email
String
UNIQUE
NOT NULL对应 SQL 大致就是：
CREATE TABLE user (
id INTEGER PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(200) NOT NULL UNIQUE
);ORM 的意义就非常直观了：
class User(db.Model)其实是在声明：
我要定义一张数据库表对应的 Python model。
10. 什么叫 Primary Key？
这里：
primary_key=True非常重要。
Primary Key：
唯一标识一行数据。
例如：
id | name
---+------
1 | Alice
2 | Bob
3 | Alice名字可以重复。
但：
id不能重复。
所以：
id = 1永远明确指向某一个用户。
因此 REST API 经常设计成：
GET /users/1
GET /users/211. 创建数据库表
定义 model 不代表数据库中已经真的存在这张表。
需要：
with app.app_context():
db.create_all()完整：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
db = SQLAlchemy(app)
class User(db.Model):
id = db.Column(
db.Integer,
primary_key=True
)
name = db.Column(
db.String(100),
nullable=False
)
email = db.Column(
db.String(200),
unique=True,
nullable=False
)
with app.app_context():
db.create_all()
if __name__ == "__main__":
app.run(debug=True)create_all() 会根据已经定义的 models 创建缺失的表；Flask-SQLAlchemy 文档也要求这类数据库操作在 Flask application context 中执行。(Flask-SQLAlchemy)
运行：
python app.py就会产生 SQLite 数据库。
12. Create：插入数据
现在来看 CRUD 第一项。
创建用户：
user = User(
name="Alice",
email="alice@test.com"
)注意：
现在只是创建了一个普通 Python object。
还没有进入数据库：
Python
User(...)↓
db.session.add(user)↓
告诉 SQLAlchemy：
我打算把它写入数据库。
最后：
db.session.commit()才真正提交事务。
所以：
user = User(
name="Alice",
email="alice@test.com"
)
db.session.add(user)
db.session.commit()可以理解成：
new Python object
↓
session.add()
↓
pending
↓
commit()
↓
databaseSQLAlchemy ORM 官方也是通过 Session.add() 和 transaction commit 来持久化对象。(SQLAlchemy 文档)
13. 把它做成 REST API
我们已经知道：
POST /api/users表示创建 User。
因此：
from flask import request
@app.route("/api/users", methods=["POST"])
def create_user():
data = request.get_json()
user = User(
name=data["name"],
email=data["email"]
)
db.session.add(user)
db.session.commit()
return {
"id": user.id,
"name": user.name,
"email": user.email
}, 201客户端：
curl -X POST \
http://127.0.0.1:5000/api/users \
-H "Content-Type: application/json" \
-d '{
"name": "Alice",
"email": "alice@test.com"
}'数据库变成：
id | name | email
---+-------+----------------
1 | Alice | alice@test.com返回：
{
"id": 1,
"name": "Alice",
"email": "alice@test.com"
}14. Read：查询数据
我们希望：
GET /api/users/1查数据库。
现代 SQLAlchemy 2.x 推荐使用 select() + session，而旧教程常见的 Model.query / session.query() 已属于 legacy query interface。(Flask-SQLAlchemy)
Flask-SQLAlchemy 提供一个方便的方法：
user = db.session.get(User, user_id)例如：
@app.route("/api/users/<int:user_id>")
def get_user(user_id):
user = db.session.get(User, user_id)
if user is None:
return {
"error": "user not found"
}, 404
return {
"id": user.id,
"name": user.name,
"email": user.email
}请求：
GET /api/users/1流程：
GET /api/users/1
↓
Flask
↓
db.session.get(User, 1)
↓
database
↓
User object
↓
JSON15. 查询所有用户
使用现代 select 写法：
@app.route("/api/users")
def get_users():
users = db.session.execute(
db.select(User)
).scalars().all()
return [
{
"id": user.id,
"name": user.name,
"email": user.email
}
for user in users
]核心：
db.select(User)可以理解成：
SELECT *
FROM user;SQLAlchemy 2.x 官方查询接口也是以 select() 作为主要查询构造方式。(SQLAlchemy 文档)
16. 条件查询
例如按 email 查询：
stmt = db.select(User).where(
User.email == "alice@test.com"
)
user = db.session.execute(
stmt
).scalar_one_or_none()对应 SQL 的思想：
SELECT *
FROM user
WHERE email = 'alice@test.com';所以你可以开始理解 ORM 并不是魔法：
db.select(User).where(
User.email == email
)本质是在构造：
SELECT ...
WHERE email = ...17. Update：修改数据
API：
PATCH /api/users/1例如：
{
"name": "Alice Smith"
}Flask：
@app.route(
"/api/users/<int:user_id>",
methods=["PATCH"]
)
def update_user(user_id):
user = db.session.get(User, user_id)
if user is None:
return {
"error": "user not found"
}, 404
data = request.get_json()
if "name" in data:
user.name = data["name"]
if "email" in data:
user.email = data["email"]
db.session.commit()
return {
"id": user.id,
"name": user.name,
"email": user.email
}注意这里没有：
db.session.add(user)因为 user 本身已经由当前 Session 加载并被 Session 管理。
你修改：
user.name = ...然后：
db.session.commit()SQLAlchemy 会把相应变化同步到数据库。
18. Delete：删除
@app.route(
"/api/users/<int:user_id>",
methods=["DELETE"]
)
def delete_user(user_id):
user = db.session.get(User, user_id)
if user is None:
return {
"error": "user not found"
}, 404
db.session.delete(user)
db.session.commit()
return "", 204核心：
db.session.delete(user)
db.session.commit()对应：
DELETE FROM user
WHERE id = 1;19. 现在 CRUD 已经完整了
整个映射关系是：
HTTP API SQL / Database
POST /users
↓
User(...)
session.add()
commit()
INSERT
GET /users/1
↓
session.get()
SELECT
PATCH /users/1
↓
user.name = ...
commit()
UPDATE
DELETE /users/1
↓
session.delete()
commit()
DELETE这张图非常值得记住。
20. db.session 到底是什么？
这是数据库部分最值得理解的概念之一。
不要把：
db.session理解成“数据库”。
更准确来说，它是：
Python 对象世界和数据库 transaction 之间的工作上下文。
例如：
alice = User(...)
bob = User(...)
db.session.add(alice)
db.session.add(bob)此时 Session 记录：
准备 INSERT Alice
准备 INSERT Bob然后：
db.session.commit()统一提交。
SQLAlchemy 的 Session 同时承担 ORM 对象状态管理和 transaction 交互等职责。(SQLAlchemy 文档)
21. 为什么必须 commit()？
假设银行转账：
Alice -100
Bob +100你显然不希望发生：
Alice -100 成功
Bob +100 失败数据库 transaction 希望：
两个操作全部成功或者：
全部失败所以：
alice.balance -= 100
bob.balance += 100
db.session.commit()把它们作为一个 transaction 提交。
如果中间出现异常：
try:
...
db.session.commit()
except Exception:
db.session.rollback()就可以回滚 transaction。
这是数据库非常核心的概念。
22. 完整可运行版本
把我们现在的东西全部放进去：
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
db = SQLAlchemy(app)
class User(db.Model):
id = db.Column(
db.Integer,
primary_key=True
)
name = db.Column(
db.String(100),
nullable=False
)
email = db.Column(
db.String(200),
unique=True,
nullable=False
)
@app.route("/api/users", methods=["POST"])
def create_user():
data = request.get_json()
user = User(
name=data["name"],
email=data["email"]
)
db.session.add(user)
db.session.commit()
return {
"id": user.id,
"name": user.name,
"email": user.email
}, 201
@app.route("/api/users", methods=["GET"])
def get_users():
users = db.session.execute(
db.select(User)
).scalars().all()
return [
{
"id": user.id,
"name": user.name,
"email": user.email
}
for user in users
]
@app.route("/api/users/<int:user_id>")
def get_user(user_id):
user = db.session.get(User, user_id)
if user is None:
return {
"error": "user not found"
}, 404
return {
"id": user.id,
"name": user.name,
"email": user.email
}
@app.route(
"/api/users/<int:user_id>",
methods=["PATCH"]
)
def update_user(user_id):
user = db.session.get(User, user_id)
if user is None:
return {
"error": "user not found"
}, 404
data = request.get_json()
if "name" in data:
user.name = data["name"]
if "email" in data:
user.email = data["email"]
db.session.commit()
return {
"id": user.id,
"name": user.name,
"email": user.email
}
@app.route(
"/api/users/<int:user_id>",
methods=["DELETE"]
)
def delete_user(user_id):
user = db.session.get(User, user_id)
if user is None:
return {
"error": "user not found"
}, 404
db.session.delete(user)
db.session.commit()
return "", 204
with app.app_context():
db.create_all()
if __name__ == "__main__":
app.run(debug=True)这已经是一个真正有持久化能力的 REST API 了。
23. 和 Blueprint 结合
上一节我们刚学了 Blueprint。
现在项目就可以拆成：
project/
│
├── app.py
│
├── extensions.py
│
├── models/
│ └── user.py
│
└── routes/
└── users.py这里一个很常见的模式是：
extensions.py
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()models/user.py
from extensions import db
class User(db.Model):
id = db.Column(
db.Integer,
primary_key=True
)
name = db.Column(
db.String(100),
nullable=False
)routes/users.py
from flask import Blueprint
from extensions import db
from models.user import User
users_bp = Blueprint(
"users",
__name__,
url_prefix="/api/users"
)
@users_bp.route("/")
def get_users():
users = db.session.execute(
db.select(User)
).scalars().all()
return [
{
"id": user.id,
"name": user.name
}
for user in users
]app.py
from flask import Flask
from extensions import db
from routes.users import users_bp
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = \
"sqlite:///app.db"
db.init_app(app)
app.register_blueprint(users_bp)
with app.app_context():
db.create_all()
if __name__ == "__main__":
app.run(debug=True)Flask-SQLAlchemy 官方支持这种先创建 db、之后通过 init_app() 绑定 application 的模式，这也更适合后续 Application Factory。(Flask-SQLAlchemy)
24. 到这里你已经学到了什么？
现在把前面整个 Flask 路线串起来：
Browser / Frontend
│
│ HTTP
▼
CORS
│
▼
Flask App
│
▼
Blueprint
│
▼
Route
│
▼
SQLAlchemy
│
▼
SQLite各自职责是：
Flask
HTTP → Python
REST
API 怎么设计
CORS
哪些浏览器 Origin 可以访问
Blueprint
API 如何模块化
SQLAlchemy
Python 如何操作数据库
SQLite
真正保存数据25. 数据库这里真正需要掌握的核心知识
现在先不要急着学很多高级 SQLAlchemy API。
第一阶段只要真正理解：
Table
Row
Column
Primary Key
SQL:
SELECT
INSERT
UPDATE
DELETE
ORM:
Model
Session
CRUD:
Create
Read
Update
Delete
Transaction:
commit
rollback尤其是：
Python object
↕
SQLAlchemy ORM
↕
SQL
↕
Database这个关系一定要清楚。
接下来数据库部分还有两个非常重要的主题
我建议继续按照：
1. 表之间的关系
Primary Key
Foreign Key
One-to-Many
Many-to-Many
JOIN
relationship()
↓
2. 数据库 Migration
Flask-Migrate
Alembic来学。
其中 表关系 + Foreign Key + JOIN 比 migration 更值得先学。
因为真正的应用很快就会出现：
User
│
└── Orders
Order
│
└── Products而这正是关系数据库最核心的地方。