Python SQLite 实战：参数化查询、事务回滚与数据边界

# 不要这样写：输入可能改变 SQL 的语义。
sql = f"SELECT id, title FROM notes WHERE title = '{title}'"

row = connection.execute(
    "SELECT id, title FROM notes WHERE title = ?",
    (title,),
).fetchone()

SORT_COLUMNS = {
    "newest": "id",
    "title": "title",
}
SORT_DIRECTIONS = {"asc": "ASC", "desc": "DESC"}


def list_notes(connection, *, sort="newest", direction="desc", limit=20, offset=0):
    if not isinstance(sort, str) or not isinstance(direction, str):
        raise ValueError("排序参数必须是字符串")
    column = SORT_COLUMNS.get(sort)
    order = SORT_DIRECTIONS.get(direction.lower())
    if column is None or order is None:
        raise ValueError("不支持的排序方式")
    if type(limit) is not int or type(offset) is not int or limit < 0 or offset < 0:
        raise ValueError("分页参数必须是非负整数")

    sql = (
        "SELECT id, title, body, status FROM notes "
        f"ORDER BY {column} {order}, id ASC LIMIT ? OFFSET ?"
    )
    return list(connection.execute(sql, (limit, offset)))

from contextlib import closing
import sqlite3


def initialize(connection):
    connection.execute("PRAGMA foreign_keys = ON")
    connection.execute(
        """
        CREATE TABLE notes (
            id INTEGER PRIMARY KEY,
            title TEXT NOT NULL CHECK (length(title) > 0),
            body TEXT NOT NULL,
            status TEXT NOT NULL CHECK (status IN ('draft', 'published'))
        )
        """
    )
    connection.execute(
        """
        CREATE TABLE note_events (
            id INTEGER PRIMARY KEY,
            note_id INTEGER NOT NULL REFERENCES notes(id),
            event_type TEXT NOT NULL CHECK (event_type IN ('created', 'published'))
        )
        """
    )


def create_note_with_event(connection, title, body, event_type="created"):
    normalized_title = title.strip() if isinstance(title, str) else ""
    if not normalized_title:
        raise ValueError("标题不能为空")
    if not isinstance(body, str):
        raise TypeError("正文必须是字符串")

    # 这个函数拥有自己的事务边界；不要再从更外层事务中嵌套调用。
    with connection:
        cursor = connection.execute(
            "INSERT INTO notes(title, body, status) VALUES (?, ?, ?)",
            (normalized_title, body, "draft"),
        )
        note_id = cursor.lastrowid
        connection.execute(
            "INSERT INTO note_events(note_id, event_type) VALUES (?, ?)",
            (note_id, event_type),
        )
    return int(note_id)


with closing(sqlite3.connect(":memory:", timeout=2.0)) as connection:
    connection.row_factory = sqlite3.Row
    initialize(connection)
    note_id = create_note_with_event(connection, " SQLite 入门 ", "先写测试，再做修改。")
    row = connection.execute(
        "SELECT id, title, status FROM notes WHERE id = ?",
        (note_id,),
    ).fetchone()
    print(dict(row))

import sqlite3
import unittest
from contextlib import closing


class SQLiteGuideTests(unittest.TestCase):
    def setUp(self):
        self.connection_context = closing(sqlite3.connect(":memory:"))
        self.connection = self.connection_context.__enter__()
        self.connection.row_factory = sqlite3.Row
        initialize(self.connection)

    def tearDown(self):
        self.connection_context.__exit__(None, None, None)

    def test_sql_looking_title_is_stored_as_data(self):
        title = "x'); DROP TABLE notes; --"
        note_id = create_note_with_event(self.connection, title, "正文")
        row = self.connection.execute(
            "SELECT title FROM notes WHERE id = ?",
            (note_id,),
        ).fetchone()
        self.assertEqual(row["title"], title)
        self.assertEqual(
            self.connection.execute("SELECT COUNT(*) FROM notes").fetchone()[0],
            1,
        )

    def test_failed_second_write_rolls_back_the_first(self):
        with self.assertRaises(sqlite3.IntegrityError):
            create_note_with_event(self.connection, "临时草稿", "正文", "unknown")

        self.assertEqual(
            self.connection.execute("SELECT COUNT(*) FROM notes").fetchone()[0],
            0,
        )
        self.assertEqual(
            self.connection.execute("SELECT COUNT(*) FROM note_events").fetchone()[0],
            0,
        )

    def test_sorting_uses_allowlisted_columns_and_bound_pagination(self):
        for title in ("C", "A", "B"):
            create_note_with_event(self.connection, title, "正文")

        rows = list_notes(self.connection, sort="title", direction="asc", limit=2, offset=1)
        self.assertEqual([row["title"] for row in rows], ["B", "C"])

    def test_unknown_or_malformed_sort_and_negative_page_size_are_rejected(self):
        for sort in ("title; DROP TABLE notes", [], None):
            with self.subTest(sort=sort), self.assertRaises(ValueError):
                list_notes(self.connection, sort=sort)
        with self.assertRaises(ValueError):
            list_notes(self.connection, direction=[])
        with self.assertRaises(ValueError):
            list_notes(self.connection, limit=-1)


if __name__ == "__main__":
    unittest.main()

python -m unittest -v sqlite_guide.py