2026年8月31日 星期一

Python一下：用SQLite建立水井村場域、感測器與巡查資料庫

《Python一下：從風土資料到智慧生活》第 25 篇

用SQLite建立水井村場域、感測器與巡查資料庫

把散落的CSV整理成有關聯、有約束、可查詢，也能安全備份的場域資料。

CH13 關聯式資料庫SQLite參數化SQL
學習目標
完成本篇後，你能解釋資料表、列、欄位、主鍵與外鍵，使用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("交易失敗，已回復本次變更。")
        raise

with 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設備，加入正常、未知及無效讀值，查詢各場域的資料品質數量。
挑戰B｜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場域資料庫的底線。
學習銜接
本篇完成SQLite單機資料庫，下一篇將先學會SQL查詢語法，再進一步連接MySQL與多人系統。

下一篇：《Python一下：從SELECT到JOIN——用SQL回答智慧養殖問題》
張貼者： 於
標籤： , ,

沒有留言:

張貼留言

訂閱： 張貼留言 (Atom)