《Python一下：從風土資料到智慧生活》第 25 篇
用SQLite建立水井村場域、感測器與巡查資料庫
把散落的CSV整理成有關聯、有約束、可查詢，也能安全備份的場域資料。
學習目標
完成本篇後，你能解釋資料表、列、欄位、主鍵與外鍵，使用Python內建
完成本篇後，你能解釋資料表、列、欄位、主鍵與外鍵，使用Python內建
sqlite3 建立資料庫及關聯表，以參數化SQL新增、查詢、更新與刪除資料，使用交易維持一致性，並建立安全備份。一、CSV不夠用的時候
CSV適合交換資料，但多座場域、感測器與大量巡查紀錄之間具有關係。若每列重複場域名稱和設備資訊，容易出現拼字不一、重複資料及刪改不同步。關聯式資料庫以表格和鍵值建立一致關係。
| 概念 | 本篇例子 |
|---|---|
| 資料表 | sites、sensors、readings |
| 主鍵 | 每個場域或設備的唯一id |
| 外鍵 | 感測器所屬場域、讀值所屬感測器 |
| 約束 | 名稱不可空白、pH基本理論範圍 |
資料治理提醒：教材使用「示範池A～G」匿名名稱，不保存真實養殖戶姓名、電話、精確座標、帳密或設備Token。資料庫檔案本身沒有自動等同完善權限控制；真實系統仍需最小權限、加密、備份、存取紀錄與資料保存政策。
二、連線並啟用外鍵
import sqlite3
from pathlib import Path
def connect_database(path):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(path, timeout=10)
connection.execute("PRAGMA foreign_keys = ON")
connection.row_factory = sqlite3.Row
return connection
connection = connect_database("data/shuijing_demo.db")
print(connection.execute(
"PRAGMA foreign_keys"
).fetchone()[0])SQLite外鍵檢查應在每個連線明確啟用。Row 讓查詢結果可用欄位名稱讀取。
三、設計三張關聯表
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS sites (
site_id INTEGER PRIMARY KEY,
site_code TEXT NOT NULL UNIQUE,
display_name TEXT NOT NULL,
active INTEGER NOT NULL DEFAULT 1
CHECK (active IN (0, 1))
);
CREATE TABLE IF NOT EXISTS sensors (
sensor_id INTEGER PRIMARY KEY,
sensor_code TEXT NOT NULL UNIQUE,
site_id INTEGER NOT NULL,
kind TEXT NOT NULL,
unit TEXT NOT NULL,
FOREIGN KEY (site_id) REFERENCES sites(site_id)
ON UPDATE CASCADE ON DELETE RESTRICT
);
CREATE TABLE IF NOT EXISTS readings (
reading_id INTEGER PRIMARY KEY,
sensor_id INTEGER NOT NULL,
observed_at TEXT NOT NULL,
value REAL,
quality TEXT NOT NULL
CHECK (quality IN ('valid', 'unknown', 'invalid')),
note TEXT NOT NULL DEFAULT '',
FOREIGN KEY (sensor_id) REFERENCES sensors(sensor_id)
ON DELETE RESTRICT,
UNIQUE (sensor_id, observed_at)
);
"""
with connection:
connection.executescript(SCHEMA_SQL)value 允許NULL，因為設備離線時「未知」不等於0；同時用quality標記資料品質。真實門檻不寫死在這個通用資料表。
四、新增七座匿名場域
sites = [
(f"SITE-{number:02d}", f"示範池{letter}")
for number, letter in enumerate("ABCDEFG", start=1)
]
with connection:
connection.executemany(
"""
INSERT INTO sites (site_code, display_name)
VALUES (?, ?)
ON CONFLICT(site_code) DO NOTHING
""",
sites,
)
print("場域筆數：", connection.execute(
"SELECT COUNT(*) FROM sites"
).fetchone()[0])問號是參數佔位符。資料值永遠透過第二個參數傳入，不用f-string或字串拼接產生SQL。
五、新增感測器並取得場域主鍵
def find_site_id(connection, site_code):
row = connection.execute(
"SELECT site_id FROM sites WHERE site_code = ?",
(site_code,),
).fetchone()
if row is None:
raise ValueError(f"找不到場域：{site_code}")
return row["site_id"]
site_id = find_site_id(connection, "SITE-01")
with connection:
connection.execute(
"""
INSERT INTO sensors
(sensor_code, site_id, kind, unit)
VALUES (?, ?, ?, ?)
ON CONFLICT(sensor_code) DO NOTHING
""",
("TEMP-01", site_id, "temperature", "°C"),
)六、為何不能拼接SQL？
# 錯誤示範：不要把外部文字拼進SQL
unsafe_name = "示範池A"
unsafe_sql = (
"SELECT * FROM sites WHERE display_name = '"
+ unsafe_name
+ "'"
)
# 正確方式：SQL與資料值分開
safe_row = connection.execute(
"SELECT * FROM sites WHERE display_name = ?",
(unsafe_name,),
).fetchone()
print(dict(safe_row) if safe_row else "找不到")參數化SQL可防止資料被解讀成SQL語法，也能正確處理引號。表名、欄名不能用一般值參數代替；若需要動態欄位，必須從程式內固定允許清單選擇。
七、驗證後新增讀值
from datetime import datetime
def insert_reading(
connection, sensor_code, observed_at,
value, quality, note=""
):
datetime.fromisoformat(observed_at)
if quality not in {"valid", "unknown", "invalid"}:
raise ValueError("不支援的quality")
if quality == "unknown":
value = None
elif value is None:
raise ValueError("valid或invalid必須保留原始數值")
else:
value = float(value)
with connection:
connection.execute(
"""
INSERT INTO readings
(sensor_id, observed_at, value, quality, note)
SELECT sensor_id, ?, ?, ?, ?
FROM sensors
WHERE sensor_code = ?
""",
(
observed_at, value, quality,
str(note), sensor_code,
),
)
insert_reading(
connection, "TEMP-01",
"2026-08-31T08:00:00+08:00",
27.2, "valid",
)八、確認INSERT真的找到感測器
前一個 INSERT...SELECT 若找不到sensor_code，可能插入0列而不報錯。可檢查游標的 rowcount：
def insert_unknown(connection, sensor_code, observed_at):
with connection:
cursor = connection.execute(
"""
INSERT INTO readings
(sensor_id, observed_at, value, quality, note)
SELECT sensor_id, ?, NULL, 'unknown', ?
FROM sensors
WHERE sensor_code = ?
""",
(observed_at, "設備離線", sensor_code),
)
if cursor.rowcount != 1:
raise ValueError(f"找不到感測器：{sensor_code}")九、JOIN查詢巡查紀錄
rows = connection.execute(
"""
SELECT
s.site_code,
s.display_name,
e.sensor_code,
e.kind,
r.observed_at,
r.value,
r.quality
FROM readings AS r
JOIN sensors AS e ON e.sensor_id = r.sensor_id
JOIN sites AS s ON s.site_id = e.site_id
ORDER BY r.observed_at DESC, s.site_code
"""
).fetchall()
for row in rows:
print(dict(row))十、彙整有效資料
summary = connection.execute(
"""
SELECT
s.site_code,
COUNT(r.reading_id) AS valid_count,
ROUND(AVG(r.value), 2) AS average_value
FROM sites AS s
LEFT JOIN sensors AS e ON e.site_id = s.site_id
LEFT JOIN readings AS r
ON r.sensor_id = e.sensor_id
AND r.quality = 'valid'
GROUP BY s.site_id, s.site_code
ORDER BY s.site_code
"""
).fetchall()
for row in summary:
print(dict(row))只對quality為valid的資料取平均，並同時呈現有效筆數。沒有有效資料時AVG為NULL，不應自動改成0。
十一、交易：全部成功或全部取消
def rename_site_and_sensor(
connection, site_code, new_name,
sensor_code, new_unit
):
try:
with connection:
connection.execute(
"""
UPDATE sites SET display_name = ?
WHERE site_code = ?
""",
(new_name, site_code),
)
connection.execute(
"""
UPDATE sensors SET unit = ?
WHERE sensor_code = ?
""",
(new_unit, sensor_code),
)
except sqlite3.DatabaseError:
print("交易失敗，已回復本次變更。")
raisewith connection 正常結束時提交，出現例外時回復。批次匯入應以合理大小分批，不要每一列各自提交。
十二、更新與刪除要先預覽
def preview_inactive_sites(connection):
return connection.execute(
"""
SELECT site_id, site_code, display_name
FROM sites
WHERE active = 0
ORDER BY site_code
"""
).fetchall()
for row in preview_inactive_sites(connection):
print("待確認：", dict(row))
# 教材預設不執行DELETE
# 刪除前還要檢查外鍵、備份、權限與保存政策十三、索引與查詢計畫
with connection:
connection.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_readings_sensor_time
ON readings(sensor_id, observed_at)
"""
)
plan = connection.execute(
"""
EXPLAIN QUERY PLAN
SELECT * FROM readings
WHERE sensor_id = ? AND observed_at >= ?
""",
(1, "2026-08-01"),
).fetchall()
for row in plan:
print(tuple(row))索引可加速特定查詢，但會增加儲存與寫入成本。應依實際查詢及資料量建立，不是每個欄位都加索引。
十四、使用SQLite備份API
import sqlite3
from pathlib import Path
def backup_database(source_connection, target_path):
target_path = Path(target_path)
if target_path.exists():
raise FileExistsError(f"備份已存在：{target_path}")
target_path.parent.mkdir(parents=True, exist_ok=True)
target = sqlite3.connect(target_path)
try:
source_connection.backup(target)
finally:
target.close()
return target_path
# backup_database(connection, "backup/shuijing_demo_v1.db")不要在資料庫正在寫入時直接複製檔案當作唯一備份流程。SQLite備份API可在連線層建立一致副本；備份後仍要驗證並做還原演練。
十五、完整性檢查與關閉連線
foreign_key_errors = connection.execute(
"PRAGMA foreign_key_check"
).fetchall()
integrity = connection.execute(
"PRAGMA integrity_check"
).fetchone()[0]
print("外鍵錯誤：", len(foreign_key_errors))
print("完整性：", integrity)
connection.close()十六、常見錯誤檢查表
| 現象 | 原因 | 修正 |
|---|---|---|
| 刪除場域後留下孤兒設備 | 未啟用外鍵 | 每個連線PRAGMA foreign_keys=ON |
| 含引號資料造成錯誤或注入 | 拼接SQL | 所有值使用參數化SQL |
| 設備離線被記為0 | 混淆未知與零值 | value=NULL並記錄quality |
| 批次中途留下半套資料 | 未使用交易 | 用with connection包住整批 |
| 備份檔無法還原 | 未驗證與演練 | 完整性檢查並測試還原 |
十七、USR實作挑戰
挑戰A｜七座場域模型
為七座匿名教學場域各建立水溫與pH設備，加入正常、未知及無效讀值，查詢各場域的資料品質數量。
為七座匿名教學場域各建立水溫與pH設備，加入正常、未知及無效讀值，查詢各場域的資料品質數量。
挑戰B｜CSV匯入交易
讀取上一章CSV，整批驗證後寫入；故意加入重複時間，確認交易能回復且錯誤可追查。
讀取上一章CSV，整批驗證後寫入；故意加入重複時間，確認交易能回復且錯誤可追查。
挑戰C｜權限與公開版
設計內部資料庫與公開摘要的欄位差異，公開版不得包含養殖戶、精確位置與設備識別資訊。
設計內部資料庫與公開摘要的欄位差異，公開版不得包含養殖戶、精確位置與設備識別資訊。
十八、與生成式AI協作
請擔任Python sqlite3與USR資料治理助教。
請檢查我的場域資料庫程式：
1. 啟用外鍵並設計主鍵、唯一鍵與CHECK；
2. 所有資料值必須參數化，不可拼接SQL；
3. 明確區分0、NULL、unknown與invalid；
4. 批次寫入使用交易，失敗時完整回復；
5. UPDATE與DELETE先預覽，教材預設不刪除；
6. 不保存個資、精確位置、密碼或Token；
7. 使用SQLite backup API並驗證還原。
請提供測試資料與失敗情境。本篇小結
關聯式資料庫不只是把CSV搬進表格，而是透過主鍵、外鍵、約束、交易與參數化查詢維持資料品質。未知值不造假、敏感資料最小化、備份可還原，才是USR場域資料庫的底線。
關聯式資料庫不只是把CSV搬進表格，而是透過主鍵、外鍵、約束、交易與參數化查詢維持資料品質。未知值不造假、敏感資料最小化、備份可還原，才是USR場域資料庫的底線。
