一、前言 #
嗨,大家好!我是拍拍君 🎉
你有沒有遇過這種情況:你的 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 sqlite3 官方文件
- SQLite 官方網站
- Python dataclasses 完全攻略 — 搭配 sqlite3 封裝資料
- Python pathlib 完全攻略 — 管理資料庫檔案路徑
- Python typing 完全攻略 — 讓你的資料庫程式碼更安全