Python 資料庫操作入門：SQLite、CRUD 與防 SQL Injection

你的 Python 程式跑起來很順，但資料存在哪裡？如果答案是「存成一個 JSON 檔案」，是時候認識資料庫了。這篇帶你用 Python 內建的 SQLite 建立資料庫、寫基本的 CRUD 指令，學會用參數化查詢擋掉最常見的資安漏洞 SQL Injection，最後看懂 ORM（SQLAlchemy）到底在幫你做什麼。

跟 AI 一起寫資料庫程式時，最常出現的三個雷是忘記 commit、忘記關閉連線、還有用字串拼接組 SQL。學完這篇，你會有一句隨時能用的口訣，一眼看穿 AI 寫出來的 SQL 有沒有問題。

SQLite 連線與基本操作

SQLite 是 Python 內建的輕量資料庫，不需要另外安裝伺服器，一個 .db 檔就是一個完整的資料庫，最適合學習和小型專案。

import sqlite3

# 連線：檔案不存在會自動建立
conn = sqlite3.connect("students.db")
cursor = conn.cursor()

# 建立資料表，IF NOT EXISTS 代表表已存在時不會報錯
cursor.execute("""
    CREATE TABLE IF NOT EXISTS students (
        id    INTEGER PRIMARY KEY AUTOINCREMENT,
        name  TEXT    NOT NULL,
        score INTEGER DEFAULT 0
    )
""")
conn.commit()    # 寫入操作一定要 commit
conn.close()

# 推薦寫法：用 with 自動管理連線
with sqlite3.connect("students.db") as conn:
    conn.row_factory = sqlite3.Row   # 讓查詢結果可用欄位名稱存取
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM students")
    rows = cursor.fetchall()

SQL 四大指令：CRUD

with sqlite3.connect("students.db") as conn:
    cursor = conn.cursor()

    # CREATE：新增，用問號參數化，防止 Injection
    cursor.execute(
        "INSERT INTO students (name, score) VALUES (?, ?)",
        ("Alice", 92)
    )
    conn.commit()   # 寫入一定要 commit

    # READ：查詢
    cursor.execute("SELECT * FROM students WHERE score >= ?", (60,))
    rows = cursor.fetchall()

    # UPDATE：更新
    cursor.execute(
        "UPDATE students SET score = ? WHERE name = ?",
        (95, "Alice")
    )
    conn.commit()

    # DELETE：刪除
    cursor.execute("DELETE FROM students WHERE score < ?", (60,))
    conn.commit()

    # 批次新增：executemany
    data = [("Bob", 78), ("Carol", 85)]
    cursor.executemany(
        "INSERT INTO students (name, score) VALUES (?, ?)", data
    )
    conn.commit()

參數化查詢：防 SQL Injection

SQL Injection 是讓攻擊者能夠輸入惡意 SQL 指令的漏洞。參數化查詢（用問號佔位符）是唯一正確的防護方式，AI 生成的初版程式常用字串拼接，你必須看得出來並要求修正。

name = input("name: ")
sql = "SELECT * FROM students WHERE name = '" + name + "'"
cursor.execute(sql)

# 輸入 ' OR '1'='1 會查出所有資料
# 輸入 '; DROP TABLE students; -- 會直接刪除整張表

name = input("name: ")
cursor.execute("SELECT * FROM students WHERE name = ?", (name,))

ORM 概念：SQLAlchemy 入門

ORM（Object-Relational Mapping）讓你用 Python 物件操作資料庫，不需要手寫 SQL。一個 class 對應一張表，一個實例對應一筆資料。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, Session

class Base(DeclarativeBase): pass

class Student(Base):          # 一個 class 對應一張表
    __tablename__ = "students"
    id    = Column(Integer, primary_key=True)
    name  = Column(String, nullable=False)
    score = Column(Integer, default=0)

engine = create_engine("sqlite:///students.db")
Base.metadata.create_all(engine)  # 自動建表

with Session(engine) as session:
    alice = Student(name="Alice", score=92)
    session.add(alice)
    session.commit()

    results = session.query(Student).filter(
        Student.score >= 60
    ).all()
    for r in results:
        print(r.name, r.score)  # 直接用屬性存取

AI 常見資料庫陷阱

陷阱一：沒有 commit，資料沒有存入

cursor.execute("INSERT INTO students VALUES (?, ?)", ("Alice", 92))
# 沒有呼叫 conn.commit，程式結束後資料就消失

陷阱二：不用 with，連線資源洩漏

conn = sqlite3.connect("db.sqlite")
# 中間做了一堆操作
# 忘記 conn.close，長期運行耗盡連線資源
# 改用：with sqlite3.connect("db.sqlite") as conn:

陷阱三：用索引取欄位，可讀性差

row = cursor.fetchone()
print(row[0], row[2])  # 這是哪兩個欄位？

conn.row_factory = sqlite3.Row
row = cursor.fetchone()
print(row["name"], row["score"])  # 清楚可讀

三道遞進題：讀懂、改寫、抓錯

這三題分別練不同的能力：讀懂是追蹤資料從新增到查詢的完整流程；改寫是把危險的字串拼接改成參數化查詢；抓錯是一次抓出 commit 缺失、Injection 漏洞、連線未關閉這三種最常見的組合問題。

題目一：讀懂（基礎）

import sqlite3

with sqlite3.connect(":memory:") as conn:
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    cur.execute("CREATE TABLE t (name TEXT, score INT)")
    cur.executemany(
        "INSERT INTO t VALUES (?, ?)",
        [("A", 90), ("B", 55), ("C", 75)]
    )
    conn.commit()
    cur.execute(
        "SELECT * FROM t WHERE score >= ? ORDER BY score DESC",
        (60,)
    )
    for row in cur.fetchall():
        print(row["name"], row["score"])

任務：說明冒號加 memory 這種寫法、executemany、row_factory 各自的作用，以及最終輸出結果。

題目二：改寫（進階）

def search_student(conn, name):
    sql = "SELECT * FROM students WHERE name = '" + name + "'"
    cursor = conn.cursor()
    cursor.execute(sql)
    return cursor.fetchall()

題目三：抓錯（高階）

資料庫的錯誤最難發現，程式不會崩潰，但資料沒存進去、查出來是空的、或已經被注入攻擊，往往完全看不出來。

import sqlite3

conn = sqlite3.connect("app.db")
cur = conn.cursor()

def add_user(name, email):
    cur.execute(
        "INSERT INTO users VALUES ('"
        + name + "', '" + email + "')"  # 問題一
    )                                    # 問題二：少了什麼？

def get_user(uid):
    cur.execute(
        "SELECT * FROM users WHERE id=" + str(uid)  # 問題三
    )
    return cur.fetchone()
# 問題四：conn 在哪裡關閉？

任務：找出這段程式碼的四個問題，並說明各自的危險原因。

重點整理

模組六完成清單

• 能用 sqlite3 建立連線、建表、執行 CRUD 四種操作
• 知道所有寫入操作後都需要呼叫 conn.commit（）
• 能辨識並修正 SQL Injection，把字串拼接改成參數化查詢
• 能設定 row_factory，改用欄位名稱取值
• 了解 ORM 的概念和 SQLAlchemy 的基本用法
• 完成題目一：說明記憶體資料庫、executemany、輸出結果
• 完成題目二：請 AI 修正 SQL Injection 並加上例外處理
• 完成題目三：找出四個資料庫程式碼的問題