专项 · 后端第 3 课 / 共 6 课免费阅读

在代码里用数据库 · 后端开发

后端怎么安全地读写数据库？

为什么需要它

几乎所有后端都是数据库外面薄薄的一层。区分可靠代码和危险代码的，是两个习惯：参数化查询，让 SQL 注入不可能发生；事务，保证一组修改要么全部完成，要么完全不做。

直觉理解

Python 自带的 sqlite3 和所有数据库库用法一样：

db = sqlite3.connect("shop.db")
rows = db.execute("SELECT id, name FROM users WHERE name = ?", ("Ada",)).fetchall()

execute(sql, 参数) 执行一条语句。? 由数据库来填，填进去的是数据，永远不会被当成 SQL。这就是防住注入的原因：用户输入什么都改变不了查询本身。
fetchone() 返回一行（一个元组）或 None；fetchall() 返回由行组成的列表。
cursor.lastrowid 是刚插入的那一行的 id。
违反规则（UNIQUE、NOT NULL、CHECK）会抛出 sqlite3.IntegrityError：捕获它，返回 409 或 400。

事务。 转账是两次更新。如果第二次失败，第一次必须撤销。with db: 就是做这个的：代码块正常结束就提交，有异常跑出来就回滚。

定义

参数化查询

SQL with ? placeholders, and the values passed separately: execute(sql, params). The values can never become SQL.
用 ? 占位的 SQL，值单独传入：execute(sql, 参数)。这些值永远不会变成 SQL。

事务

A group of changes that is saved completely (commit) or not at all (rollback).
一组修改，要么全部保存（提交），要么完全不保存（回滚）。

示例

用占位符查询

动手试试

查回来的每一行是一个元组。

import sqlite3

def make_db():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT UNIQUE)")
    db.executemany("INSERT INTO users (name, email) VALUES (?, ?)",
                   [("Ada", "ada@example.com"), ("Linus", "linus@example.com")])
    return db

db = make_db()
row = db.execute("SELECT id, email FROM users WHERE name = ?", ("Linus",)).fetchone()
print(row)
print(db.execute("SELECT name FROM users WHERE name = ?", ("Zed",)).fetchone())
print(db.execute("SELECT COUNT(*) FROM users").fetchone()[0])

输出

(2, 'linus@example.com')
None
2

逐步执行

L10一个全新的内存数据库，有两个用户。
L11这个元组填进 ?。注意逗号：("Linus",) 是只有一个元素的元组。
L13没有这一行：fetchone() 返回 None。None

插入，并处理重复

动手试试

lastrowid 给出新 id；UNIQUE 冲突会抛出 IntegrityError。

import sqlite3

def make_db():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT UNIQUE)")
    db.executemany("INSERT INTO users (name, email) VALUES (?, ?)",
                   [("Ada", "ada@example.com"), ("Linus", "linus@example.com")])
    return db

def add_user(db, name, email):
    try:
        cur = db.execute("INSERT INTO users (name, email) VALUES (?, ?)", (name, email))
        return 201, {"id": cur.lastrowid}
    except sqlite3.IntegrityError:
        return 409, {"error": "email already used"}

db = make_db()
print(add_user(db, "Grace", "grace@example.com"))
print(add_user(db, "Ada 2", "ada@example.com"))

输出

(201, {'id': 3})
(409, {'error': 'email already used'})

逐步执行

L13新行的 id：3。
L14ada@example.com 已经存在，所以 UNIQUE 规则不满足。
L15409 Conflict：请求和已有数据冲突。

要么全做，要么不做

动手试试

第二次更新违反规则时，with db: 会把两次更新一起回滚。

import sqlite3

def make_bank():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, owner TEXT, balance INTEGER CHECK (balance >= 0))")
    db.executemany("INSERT INTO accounts (owner, balance) VALUES (?, ?)", [("Ada", 100), ("Bo", 20)])
    db.commit()
    return db

def balances(db):
    return [row[0] for row in db.execute("SELECT balance FROM accounts ORDER BY id")]

db = make_bank()
try:
    with db:
        db.execute("UPDATE accounts SET balance = balance + 50 WHERE id = 2")
        db.execute("UPDATE accounts SET balance = balance - 50 WHERE id = 1")
        db.execute("UPDATE accounts SET balance = balance - 50 WHERE id = 1")
        db.execute("UPDATE accounts SET balance = balance - 50 WHERE id = 1")
except sqlite3.IntegrityError:
    print("rolled back")
print(balances(db))

输出

rolled back
[100, 20]

逐步执行

L15开始一个事务。
L19Ada 会变成 -50：CHECK 规则失败，抛出异常。
L21异常离开了 with 代码块，所以里面的每次更新都撤销了。
L22和之前完全一样。[100, 20]

常见错误

用 f-string 拼 SQL

f"... WHERE name = '{name}'" 让用户能输入 ' OR '1'='1 来改写你的查询。永远使用 ? 占位符。

(value) 不是元组

参数必须是元组或列表：("Ada",)，带逗号。("Ada") 只是一个字符串，每个字母都会被当成一个参数。

改到一半

两个相关的更新如果不放在事务里，程序在中间停下就会留下坏数据。用 with db: 把它们放在一起。

小测验

1. 为什么 ? 占位符能防住 SQL 注入？

A数据库把值单独收下，当作数据，永远不当作 SQL
B它们会去掉输入里的引号
C它们只允许字母和数字

查看答案

答案：A

在看到任何值之前，查询的结构就已经固定了。

2. 这段代码会输出什么？

import sqlite3

def make_db():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT UNIQUE)")
    db.executemany("INSERT INTO users (name, email) VALUES (?, ?)",
                   [("Ada", "ada@example.com"), ("Linus", "linus@example.com")])
    return db

db = make_db()
rows = db.execute("SELECT name FROM users ORDER BY name DESC").fetchall()
print(rows)

A
```
[('Linus',), ('Ada',)]
```
B
```
['Linus', 'Ada']
```
C
```
[('Ada',), ('Linus',)]
```

查看答案

答案：A

fetchall 返回元组的列表，即使只有一列。

动手练习

3 道练习，分三层：先预测程序的输出，再按新需求改程序，最后从零写一个。代码在浏览器里运行，提交马上判题。

预测输出插入并计数
准确写出它打印的内容。
改需求堵上注入漏洞
find_user 用输入拼 SQL，所以 find_user(db, "x' OR '1'='1") 会返回一个本不该返回的用户。用 ? 占位符重写这个查询。
从零写转账
给 accounts 表写一个 transfer(db, from_id, to_id, amount)（表由 make_bank 创建）。
- 在事务里（with db:）把 amount 从一个账户转到另一个账户
- 如果转出方钱不够，表上的 CHECK (balance >= 0) 规则会抛出 sqlite3.IntegrityError：捕获它，确保什么都没变，返回 False
- 成功时返回 True

在互动课里练这一课

打开互动课，这些例子都能直接改、直接运行，练习题提交马上判题。免费开始，不用注册。

打开互动课课程地图

互动课按顺序解锁：前面的课还没学，会先带你到课程地图。

本课术语

术语	中文	英文定义
sqlite3	sqlite3	Python's built-in library for SQLite databases.
Parameterised query	参数化查询	SQL with ? placeholders; values are passed separately.
Transaction	事务	Changes saved all together or not at all.
IntegrityError	IntegrityError	Raised when a change breaks a table rule such as UNIQUE or CHECK.