快轉到主要內容
  1. 教學文章/

sqlite3:Python 內建輕量資料庫完全攻略

·9 分鐘· loading · loading · ·
Python Sqlite3 SQL 資料庫 Database
每日拍拍
作者
每日拍拍
科學家 X 科技宅宅
目錄
Python 學習 - 本文屬於一個選集。
§ 30: 本文

一、前言
#

嗨,大家好!我是拍拍君 🎉

你有沒有遇過這種情況:你的 Python 程式需要存一些結構化的資料,用 JSON 檔太笨重、用 CSV 查詢太慢、裝 PostgreSQL 又太殺雞用牛刀?

答案就是 sqlite3 — Python 的標準函式庫,不用安裝任何東西,import sqlite3 就有一個完整的關聯式資料庫可以用!

SQLite 的特色:

  • 零安裝 — Python 內建,開箱即用
  • 零配置 — 不需要伺服器,資料存在單一檔案
  • 完整 SQL — 支援 JOIN、子查詢、索引、交易
  • 超輕量 — 資料庫檔案就是一個 .db 檔案
  • 超快速 — 對中小型資料量,速度碾壓大多數方案

今天拍拍君帶你從零開始,完整掌握 Python sqlite3 的所有必備技巧!


二、建立資料庫與連線
#

建立記憶體資料庫(測試用)
#

import sqlite3

# 建立記憶體資料庫(程式結束就消失)
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()
print("記憶體資料庫連線成功!🧠")
cursor.close()
conn.close()

建立檔案資料庫(正式用)
#

import sqlite3

# 資料庫存在檔案裡,不存在會自動建立
conn = sqlite3.connect("pypy.db")
cursor = conn.cursor()
print("檔案資料庫連線成功!📁")
cursor.close()
conn.close()

用 context manager 自動管理
#

推薦用 with 語法,自動處理 commit 和 rollback:

import sqlite3

with sqlite3.connect("pypy.db") as conn:
    cursor = conn.cursor()
    # 做你的事...
    print("自動 commit!🎯")
# 離開 with 時,如果沒有例外就自動 commit
# 如果有例外就自動 rollback

三、建立資料表
#

基本建表
#

import sqlite3

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

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS articles (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            slug TEXT UNIQUE NOT NULL,
            content TEXT,
            views INTEGER DEFAULT 0,
            published_at TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)

    print("資料表建立成功!📋")

SQLite 的資料型別
#

SQLite 的型別系統很簡單,只有五種:

SQLite 型別 Python 型別 說明
NULL None 空值
INTEGER int 整數
REAL float 浮點數
TEXT str 文字
BLOB bytes 二進位資料

💡 SQLite 是「動態型別」的資料庫,你宣告 INTEGER 但塞 TEXT 進去也不會報錯(但拍拍君建議不要這樣做啦 😅)。


四、CRUD 操作:新增、查詢、更新、刪除
#

新增資料(Create)
#

import sqlite3

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

    # 單筆新增 — 用 ? 佔位符防止 SQL injection
    cursor.execute(
        "INSERT INTO articles (title, slug, content) VALUES (?, ?, ?)",
        ("Python 入門", "python-intro", "Python 是一門簡潔的語言..."),
    )

    # 多筆新增
    articles = [
        ("httpx 教學", "python-httpx", "httpx 是新世代 HTTP 客戶端..."),
        ("asyncio 教學", "python-asyncio", "asyncio 讓你寫非同步程式..."),
        ("pathlib 教學", "python-pathlib", "pathlib 是現代路徑處理..."),
    ]
    cursor.executemany(
        "INSERT INTO articles (title, slug, content) VALUES (?, ?, ?)",
        articles,
    )

    print(f"新增了 {cursor.rowcount} 筆資料!✨")
    # 新增了 3 筆資料!✨

⚠️ 永遠用 ? 佔位符,絕對不要用 f-string 拼 SQL!

# ❌ 危險!SQL injection 攻擊
cursor.execute(f"SELECT * FROM articles WHERE slug = '{user_input}'")

# ✅ 安全!用參數化查詢
cursor.execute("SELECT * FROM articles WHERE slug = ?", (user_input,))

查詢資料(Read)
#

import sqlite3

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

    # 查詢全部
    cursor.execute("SELECT id, title, slug FROM articles")
    rows = cursor.fetchall()
    for row in rows:
        print(f"  [{row[0]}] {row[1]} ({row[2]})")

    # 查詢單筆
    cursor.execute("SELECT * FROM articles WHERE slug = ?", ("python-httpx",))
    article = cursor.fetchone()
    if article:
        print(f"\n找到文章:{article[1]}")

    # 條件查詢
    cursor.execute(
        "SELECT title FROM articles WHERE title LIKE ?",
        ("%教學%",),
    )
    tutorials = cursor.fetchall()
    print(f"\n共有 {len(tutorials)} 篇教學文章")

用 Row 物件取代 tuple
#

用 tuple 取值要記位置(row[0]row[1]),很容易搞混。設定 row_factory 可以用欄位名稱取值:

import sqlite3

with sqlite3.connect("pypy.db") as conn:
    conn.row_factory = sqlite3.Row  # 魔法在這一行!
    cursor = conn.cursor()

    cursor.execute("SELECT * FROM articles WHERE slug = ?", ("python-httpx",))
    article = cursor.fetchone()

    # 現在可以用欄位名稱了!
    print(f"標題:{article['title']}")
    print(f"Slug:{article['slug']}")
    print(f"瀏覽數:{article['views']}")

    # 也可以轉成 dict
    article_dict = dict(article)
    print(article_dict)

拍拍君強烈建議每次都加 conn.row_factory = sqlite3.Row,程式碼可讀性直接起飛!

更新資料(Update)
#

import sqlite3

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

    # 更新單筆
    cursor.execute(
        "UPDATE articles SET views = views + 1 WHERE slug = ?",
        ("python-httpx",),
    )
    print(f"更新了 {cursor.rowcount} 筆資料")

    # 批次更新
    cursor.execute(
        "UPDATE articles SET published_at = CURRENT_TIMESTAMP WHERE published_at IS NULL"
    )
    print(f"發佈了 {cursor.rowcount} 篇文章 📰")

刪除資料(Delete)
#

import sqlite3

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

    cursor.execute("DELETE FROM articles WHERE slug = ?", ("python-intro",))
    print(f"刪除了 {cursor.rowcount} 筆資料 🗑️")

五、索引:讓查詢飛快
#

當資料量變大,沒有索引的查詢會越來越慢。加索引就像給書加目錄一樣:

import sqlite3

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

    # 建立索引
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_articles_slug ON articles(slug)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_articles_views ON articles(views)")

    # 複合索引
    cursor.execute(
        "CREATE INDEX IF NOT EXISTS idx_articles_published "
        "ON articles(published_at, views)"
    )

    print("索引建立完成!🏎️")

查看現有索引:

cursor.execute("SELECT name, sql FROM sqlite_master WHERE type = 'index'")
for name, sql in cursor.fetchall():
    print(f"  📌 {name}")
    print(f"     {sql}")

💡 slug 欄位已經有 UNIQUE 約束,SQLite 會自動建索引。但對常用的查詢條件加額外索引還是很有用。


六、交易(Transaction)處理
#

SQLite 預設是 autocommit 模式,但在 Python 的 sqlite3 模組中,預設行為比較特殊:

import sqlite3

conn = sqlite3.connect("pypy.db")
cursor = conn.cursor()

try:
    # 開始交易(DML 語句自動開始交易)
    cursor.execute(
        "INSERT INTO articles (title, slug) VALUES (?, ?)",
        ("交易測試 A", "tx-test-a"),
    )
    cursor.execute(
        "INSERT INTO articles (title, slug) VALUES (?, ?)",
        ("交易測試 B", "tx-test-b"),
    )

    # 明確提交
    conn.commit()
    print("交易提交成功!✅")

except Exception as e:
    # 出錯就全部回滾
    conn.rollback()
    print(f"交易回滾!❌ 原因:{e}")

finally:
    conn.close()

Python 3.12+ 的 autocommit 參數
#

從 Python 3.12 開始,你可以明確控制自動提交行為:

import sqlite3

# 明確關閉 autocommit(推薦!)
conn = sqlite3.connect("pypy.db", autocommit=False)

try:
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO articles (title, slug) VALUES (?, ?)",
        ("新文章", "new-article"),
    )
    conn.commit()
except:
    conn.rollback()
    raise
finally:
    conn.close()

用 with 最簡潔
#

import sqlite3

with sqlite3.connect("pypy.db") as conn:
    # with 區塊內如果沒有例外 → 自動 commit
    # 如果有例外 → 自動 rollback
    conn.execute(
        "INSERT INTO articles (title, slug) VALUES (?, ?)",
        ("自動交易", "auto-tx"),
    )
# 出了 with 就自動 commit 了!

七、進階技巧
#

自訂函式
#

你可以把 Python 函式註冊到 SQLite 中,在 SQL 裡直接呼叫:

import sqlite3
import hashlib

def md5_hash(text):
    """計算 MD5 雜湊"""
    return hashlib.md5(text.encode()).hexdigest()

with sqlite3.connect("pypy.db") as conn:
    # 註冊自訂函式:名稱、參數數量、函式
    conn.create_function("md5", 1, md5_hash)

    cursor = conn.cursor()
    cursor.execute("SELECT title, md5(title) FROM articles")
    for title, hash_val in cursor.fetchall():
        print(f"  {title}{hash_val[:8]}...")

自訂聚合函式
#

import sqlite3

class StdDev:
    """計算標準差的聚合函式"""
    def __init__(self):
        self.values = []

    def step(self, value):
        if value is not None:
            self.values.append(value)

    def finalize(self):
        if len(self.values) < 2:
            return None
        mean = sum(self.values) / len(self.values)
        variance = sum((x - mean) ** 2 for x in self.values) / (len(self.values) - 1)
        return variance ** 0.5

with sqlite3.connect("pypy.db") as conn:
    conn.create_aggregate("stddev", 1, StdDev)

    cursor = conn.cursor()
    cursor.execute("SELECT stddev(views) FROM articles")
    result = cursor.fetchone()[0]
    print(f"瀏覽數標準差:{result}")

用 executescript 批次執行
#

import sqlite3

with sqlite3.connect("pypy.db") as conn:
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS tags (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL
        );

        CREATE TABLE IF NOT EXISTS article_tags (
            article_id INTEGER REFERENCES articles(id),
            tag_id INTEGER REFERENCES tags(id),
            PRIMARY KEY (article_id, tag_id)
        );

        INSERT OR IGNORE INTO tags (name) VALUES ('Python');
        INSERT OR IGNORE INTO tags (name) VALUES ('SQL');
        INSERT OR IGNORE INTO tags (name) VALUES ('教學');
    """)
    print("批次執行完成!🎯")

八、實戰範例:打造簡易部落格資料庫
#

來整合前面學的,做一個完整的部落格資料管理系統:

import sqlite3
from dataclasses import dataclass
from datetime import datetime

@dataclass
class Article:
    id: int | None
    title: str
    slug: str
    content: str
    views: int = 0
    published_at: str | None = None
    created_at: str | None = None

class BlogDB:
    def __init__(self, db_path: str = "blog.db"):
        self.db_path = db_path
        self._init_db()

    def _get_conn(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA journal_mode=WAL")    # 提升並發效能
        conn.execute("PRAGMA foreign_keys=ON")      # 啟用外鍵約束
        return conn

    def _init_db(self):
        with self._get_conn() as conn:
            conn.executescript("""
                CREATE TABLE IF NOT EXISTS articles (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    slug TEXT UNIQUE NOT NULL,
                    content TEXT NOT NULL,
                    views INTEGER DEFAULT 0,
                    published_at TEXT,
                    created_at TEXT DEFAULT CURRENT_TIMESTAMP
                );

                CREATE INDEX IF NOT EXISTS idx_slug ON articles(slug);
                CREATE INDEX IF NOT EXISTS idx_published ON articles(published_at);
            """)

    def create(self, title: str, slug: str, content: str) -> Article:
        with self._get_conn() as conn:
            cursor = conn.execute(
                "INSERT INTO articles (title, slug, content) VALUES (?, ?, ?)",
                (title, slug, content),
            )
            return Article(
                id=cursor.lastrowid,
                title=title,
                slug=slug,
                content=content,
            )

    def get_by_slug(self, slug: str) -> Article | None:
        with self._get_conn() as conn:
            row = conn.execute(
                "SELECT * FROM articles WHERE slug = ?", (slug,)
            ).fetchone()
            if row:
                return Article(**dict(row))
            return None

    def list_articles(
        self,
        limit: int = 10,
        offset: int = 0,
        published_only: bool = True,
    ) -> list[Article]:
        with self._get_conn() as conn:
            query = "SELECT * FROM articles"
            if published_only:
                query += " WHERE published_at IS NOT NULL"
            query += " ORDER BY published_at DESC LIMIT ? OFFSET ?"
            rows = conn.execute(query, (limit, offset)).fetchall()
            return [Article(**dict(row)) for row in rows]

    def publish(self, slug: str) -> bool:
        with self._get_conn() as conn:
            cursor = conn.execute(
                "UPDATE articles SET published_at = ? WHERE slug = ? AND published_at IS NULL",
                (datetime.now().isoformat(), slug),
            )
            return cursor.rowcount > 0

    def increment_views(self, slug: str) -> int:
        with self._get_conn() as conn:
            conn.execute(
                "UPDATE articles SET views = views + 1 WHERE slug = ?",
                (slug,),
            )
            row = conn.execute(
                "SELECT views FROM articles WHERE slug = ?", (slug,)
            ).fetchone()
            return row["views"] if row else 0

    def search(self, keyword: str) -> list[Article]:
        with self._get_conn() as conn:
            rows = conn.execute(
                "SELECT * FROM articles WHERE title LIKE ? OR content LIKE ?",
                (f"%{keyword}%", f"%{keyword}%"),
            ).fetchall()
            return [Article(**dict(row)) for row in rows]

    def stats(self) -> dict:
        with self._get_conn() as conn:
            row = conn.execute("""
                SELECT
                    COUNT(*) as total,
                    COUNT(published_at) as published,
                    SUM(views) as total_views,
                    AVG(views) as avg_views
                FROM articles
            """).fetchone()
            return dict(row)

# 使用範例
def main():
    db = BlogDB(":memory:")  # 用記憶體測試

    # 新增文章
    db.create("Python 入門", "python-intro", "Python 是最棒的語言!")
    db.create("httpx 教學", "python-httpx", "httpx 超好用的 HTTP 客戶端")
    db.create("SQLite 攻略", "python-sqlite3", "不用裝資料庫也能用 SQL!")

    # 發佈文章
    db.publish("python-intro")
    db.publish("python-sqlite3")

    # 模擬瀏覽
    for _ in range(42):
        db.increment_views("python-sqlite3")
    for _ in range(10):
        db.increment_views("python-intro")

    # 列出已發佈文章
    print("📰 已發佈文章:")
    for article in db.list_articles():
        print(f"  [{article.views:>4} views] {article.title}")

    # 搜尋
    print("\n🔍 搜尋 'Python':")
    for article in db.search("Python"):
        print(f"  📄 {article.title}")

    # 統計
    print("\n📊 部落格統計:")
    stats = db.stats()
    print(f"  文章總數:{stats['total']}")
    print(f"  已發佈:{stats['published']}")
    print(f"  總瀏覽:{stats['total_views']}")
    print(f"  平均瀏覽:{stats['avg_views']:.1f}")

if __name__ == "__main__":
    main()

執行結果:

📰 已發佈文章:
  [  42 views] SQLite 攻略
  [  10 views] Python 入門

🔍 搜尋 'Python':
  📄 Python 入門
  📄 httpx 教學
  📄 SQLite 攻略

📊 部落格統計:
  文章總數:3
  已發佈:2
  總瀏覽:52
  平均瀏覽:17.3

九、效能提示與常見陷阱
#

1. 開啟 WAL 模式
#

conn.execute("PRAGMA journal_mode=WAL")

WAL(Write-Ahead Logging)模式讓讀寫可以同時進行,效能大幅提升!尤其是多執行緒應用。

2. 批次插入用 executemany + 交易
#

import sqlite3
import time

with sqlite3.connect("benchmark.db") as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS numbers (n INTEGER)")

    data = [(i,) for i in range(100_000)]

    # ❌ 慢:逐筆插入
    start = time.time()
    for d in data:
        conn.execute("INSERT INTO numbers VALUES (?)", d)
    conn.commit()
    slow = time.time() - start

    conn.execute("DELETE FROM numbers")

    # ✅ 快:批次插入
    start = time.time()
    conn.executemany("INSERT INTO numbers VALUES (?)", data)
    conn.commit()
    fast = time.time() - start

    print(f"逐筆:{slow:.2f}s vs 批次:{fast:.2f}s")
    print(f"快了 {slow/fast:.1f} 倍!🚀")

3. 不要忘記關連線
#

# ❌ 忘記關連線
conn = sqlite3.connect("pypy.db")
# ... 做完事就忘了 close

# ✅ 用 with 自動管理
with sqlite3.connect("pypy.db") as conn:
    # ... 做完事自動處理
    pass

4. 多執行緒使用 check_same_thread
#

# 如果要在多執行緒間共用連線
conn = sqlite3.connect("pypy.db", check_same_thread=False)

# 但更好的做法是每個執行緒各自建連線

結語
#

sqlite3 是 Python 內建的寶藏模組:

特性 說明
安裝 不用安裝,import sqlite3 即可
部署 資料庫就是一個檔案,cp 就能備份
效能 中小型應用綽綽有餘
SQL 完整的關聯式資料庫功能
安全 ? 參數化查詢防 SQL injection

什麼時候該用 sqlite3?

  • ✅ 個人工具、小型應用、原型開發
  • ✅ 嵌入式設備、桌面應用
  • ✅ 測試用的假資料庫
  • ✅ 資料分析的中間儲存

什麼時候該換更大的資料庫?

  • ❌ 高並發寫入(考慮 PostgreSQL)
  • ❌ 分散式系統(考慮 PostgreSQL/MySQL)
  • ❌ 超大資料量(> 數 GB)

不要小看 sqlite3 的威力——拍拍君的經驗是,大部分的 side project 和工具腳本,一個 .db 檔案就夠了。別動不動就搬出 Docker + PostgreSQL 大砲打蚊子 😏

下次想存結構化資料時,試試 import sqlite3 吧!你會發現它比你想像的強大太多了 💪


延伸閱讀
#

Python 學習 - 本文屬於一個選集。
§ 30: 本文

相關文章

pathlib:優雅處理檔案路徑的現代方式
·6 分鐘· loading · loading
Python Pathlib 檔案處理 標準庫
httpx:Python 新世代 HTTP 客戶端完全攻略
·4 分鐘· loading · loading
Python Httpx HTTP Async Requests
Python collections 模組:讓你的資料結構更強大
·5 分鐘· loading · loading
Python Collections Counter Defaultdict Deque Namedtuple
MLX 入門教學:在 Apple Silicon 上跑機器學習
·4 分鐘· loading · loading
Python Mlx Apple-Silicon Machine-Learning Deep-Learning
FastAPI:Python 最潮的 Web API 框架
·5 分鐘· loading · loading
Python Fastapi Web Api Async
Docker for Python:讓你的程式在任何地方都能跑
·6 分鐘· loading · loading
Python Docker Container Devops 部署