《Python一下：從風土資料到智慧生活》第 26 篇
從SELECT到JOIN——用SQL回答智慧養殖問題
讓資料庫不只是存資料，而能回答「哪座池需要優先巡查？」
學習目標
完成本篇後，你能使用SELECT、WHERE、ORDER BY、LIMIT、GROUP BY與HAVING查詢資料；利用INNER JOIN及LEFT JOIN連接場域、感測器與讀值；以參數化SQL產生巡查清單，並說明「異常資料」不等於「養殖異常」。
完成本篇後，你能使用SELECT、WHERE、ORDER BY、LIMIT、GROUP BY與HAVING查詢資料；利用INNER JOIN及LEFT JOIN連接場域、感測器與讀值；以參數化SQL產生巡查清單，並說明「異常資料」不等於「養殖異常」。
一、先把場域問題翻成資料問題
| 場域想知道的事 | SQL工具 |
|---|---|
| 最近有哪些有效讀值？ | SELECT、WHERE、ORDER BY、LIMIT |
| 各場域平均水溫是多少？ | GROUP BY、AVG、COUNT |
| 哪些場域目前沒有資料？ | LEFT JOIN、IS NULL |
| 哪些讀值超出教材示範門檻？ | JOIN、CASE、參數化條件 |
重要觀念：SQL查到的是「符合規則的紀錄」，不是自動完成專業診斷。感測器誤差、校正狀態、物種、成長階段、天候與養殖戶經驗都會影響判讀；系統應協助巡查，不應直接取代人的決策。
二、建立可重複練習的記憶體資料庫
以下資料全部為匿名教材資料。使用 :memory: 可在記憶體中練習，程式結束後不留下檔案。
import sqlite3
connection = sqlite3.connect(":memory:")
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")
connection.executescript("""
CREATE TABLE sites (
site_id INTEGER PRIMARY KEY,
site_code TEXT NOT NULL UNIQUE,
display_name TEXT NOT NULL
);
CREATE TABLE 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)
);
CREATE TABLE readings (
reading_id INTEGER PRIMARY KEY,
sensor_id INTEGER NOT NULL,
observed_at TEXT NOT NULL,
value REAL,
quality TEXT NOT NULL,
FOREIGN KEY (sensor_id) REFERENCES sensors(sensor_id)
);
""")三、放入水井村智慧養殖示範資料
sites = [
(1, "SITE-01", "示範池A"),
(2, "SITE-02", "示範池B"),
(3, "SITE-03", "示範池C"),
]
sensors = [
(1, "TEMP-01", 1, "temperature", "°C"),
(2, "TEMP-02", 2, "temperature", "°C"),
(3, "PH-01", 1, "ph", "pH"),
]
readings = [
(1, 1, "2026-08-31T08:00:00+08:00", 27.2, "valid"),
(2, 1, "2026-08-31T09:00:00+08:00", 29.4, "valid"),
(3, 2, "2026-08-31T08:00:00+08:00", 30.1, "valid"),
(4, 2, "2026-08-31T09:00:00+08:00", None, "unknown"),
(5, 3, "2026-08-31T09:00:00+08:00", 7.4, "valid"),
]
with connection:
connection.executemany(
"INSERT INTO sites VALUES (?, ?, ?)", sites
)
connection.executemany(
"INSERT INTO sensors VALUES (?, ?, ?, ?, ?)", sensors
)
connection.executemany(
"INSERT INTO readings VALUES (?, ?, ?, ?, ?)", readings
)四、SELECT：決定要看哪些欄位
rows = connection.execute("""
SELECT sensor_id, observed_at, value, quality
FROM readings
""").fetchall()
for row in rows:
print(dict(row))教學初期可用 SELECT * 探索資料；正式查詢最好明確列出欄位，避免表格修改後輸出跟著意外改變，也減少不必要的資料傳輸。
五、WHERE：只留下可用的水溫資料
rows = connection.execute("""
SELECT observed_at, value
FROM readings
WHERE sensor_id = ?
AND quality = ?
AND value IS NOT NULL
""", (1, "valid")).fetchall()
for row in rows:
print(row["observed_at"], row["value"])SQL判斷空值要用 IS NULL 或 IS NOT NULL，不能用 = NULL。查詢值仍採參數化方式傳入。
六、ORDER BY與LIMIT：找最新紀錄
latest = connection.execute("""
SELECT observed_at, value, quality
FROM readings
WHERE sensor_id = ?
ORDER BY observed_at DESC, reading_id DESC
LIMIT 1
""", (1,)).fetchone()
print(dict(latest) if latest else "尚無資料")本例時間使用含時區的ISO 8601字串，格式一致時可正確排序。真實系統宜統一以UTC儲存，再於顯示時轉為臺灣時間。
七、計算欄位與CASE：把數值轉成巡查提示
low, high = 26.0, 30.0
rows = connection.execute("""
SELECT observed_at, value,
CASE
WHEN quality != 'valid' THEN '資料待確認'
WHEN value < ? THEN '低於示範門檻'
WHEN value > ? THEN '高於示範門檻'
ELSE '示範範圍內'
END AS status
FROM readings
WHERE sensor_id = ?
ORDER BY observed_at
""", (low, high, 1)).fetchall()
for row in rows:
print(row["observed_at"], row["value"], row["status"])26～30°C僅為SQL教學用門檻，不代表任何物種或場域的正式養殖標準。正式規則應由養殖專業者確認，並保留版本、適用條件及修改紀錄。
八、GROUP BY：各場域有幾筆有效水溫？
rows = connection.execute("""
SELECT s.site_code,
COUNT(r.reading_id) AS valid_count,
ROUND(AVG(r.value), 2) AS average_temperature,
MIN(r.value) AS minimum_temperature,
MAX(r.value) AS maximum_temperature
FROM sites AS s
LEFT JOIN sensors AS e
ON e.site_id = s.site_id
AND e.kind = 'temperature'
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 rows:
print(dict(row))COUNT(r.reading_id) 不計NULL，因此沒有有效讀值的示範池C會得到0；其平均值仍為NULL，而不是被誤寫成0°C。
九、HAVING：篩選彙總後的結果
rows = connection.execute("""
SELECT sensor_id,
COUNT(*) AS sample_count,
ROUND(AVG(value), 2) AS average_value
FROM readings
WHERE quality = 'valid'
GROUP BY sensor_id
HAVING COUNT(*) >= ?
ORDER BY sensor_id
""", (2,)).fetchall()
for row in rows:
print(dict(row))WHERE 在分組前篩選每一列；HAVING 在分組後篩選統計結果。
十、INNER JOIN：把代碼翻成場域資訊
rows = connection.execute("""
SELECT s.display_name,
e.sensor_code,
e.kind,
r.observed_at,
r.value,
e.unit
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
WHERE r.quality = 'valid'
ORDER BY r.observed_at DESC, s.site_code
""").fetchall()
for row in rows:
print(dict(row))INNER JOIN只保留兩邊都對得上的紀錄，適合查看已產生讀值的感測器。
十一、LEFT JOIN：找出沒有資料的場域
rows = connection.execute("""
SELECT s.site_code, s.display_name
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
GROUP BY s.site_id, s.site_code, s.display_name
HAVING COUNT(r.reading_id) = 0
ORDER BY s.site_code
""").fetchall()
for row in rows:
print(row["site_code"], row["display_name"])若改用INNER JOIN，完全沒有感測器或讀值的場域會消失，反而看不見最需要確認的缺漏。
十二、子查詢：取得每支感測器的最新讀值
rows = connection.execute("""
SELECT e.sensor_code, r.observed_at, r.value, r.quality
FROM sensors AS e
JOIN readings AS r ON r.sensor_id = e.sensor_id
WHERE r.reading_id = (
SELECT r2.reading_id
FROM readings AS r2
WHERE r2.sensor_id = e.sensor_id
ORDER BY r2.observed_at DESC, r2.reading_id DESC
LIMIT 1
)
ORDER BY e.sensor_code
""").fetchall()
for row in rows:
print(dict(row))用reading_id作第二排序條件，可在時間相同時得到穩定結果。若資料量很大，可再學習視窗函式與複合索引。
十三、綜合應用：產生優先巡查清單
def patrol_list(connection, low, high):
return connection.execute("""
SELECT s.site_code,
s.display_name,
e.sensor_code,
r.observed_at,
r.value,
r.quality,
CASE
WHEN r.quality != 'valid' THEN 1
WHEN r.value < ? OR r.value > ? THEN 2
ELSE 3
END AS priority
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
WHERE e.kind = 'temperature'
AND r.observed_at = (
SELECT MAX(r2.observed_at)
FROM readings AS r2
WHERE r2.sensor_id = e.sensor_id
)
ORDER BY priority, s.site_code
""", (low, high)).fetchall()
for item in patrol_list(connection, 26.0, 30.0):
print(dict(item))結果怎麼讀？優先級1表示資料未知或品質待確認，應先檢查供電、網路、探頭與校正；優先級2才是有效數值超出教材門檻。兩者需要不同的巡查行動。
十四、安全地讓使用者選擇排序
SQL參數只能替代「資料值」，不能直接替代欄名。動態排序應使用固定允許清單。
ALLOWED_SORTS = {
"time": "r.observed_at DESC",
"value": "r.value DESC",
"site": "s.site_code ASC",
}
def query_readings(connection, sort_name="time"):
order_sql = ALLOWED_SORTS.get(sort_name)
if order_sql is None:
raise ValueError("不支援的排序方式")
sql = f"""
SELECT s.site_code, r.observed_at, r.value
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
WHERE r.quality = ?
ORDER BY {order_sql}
"""
return connection.execute(sql, ("valid",)).fetchall()
print([dict(row) for row in query_readings(connection, "site")])這裡的f-string只插入程式內已核准的SQL片段；外部輸入仍不能直接放進SQL。
十五、看懂查詢計畫與索引
connection.execute("""
CREATE INDEX idx_readings_sensor_time
ON readings(sensor_id, observed_at DESC)
""")
plan = connection.execute("""
EXPLAIN QUERY PLAN
SELECT reading_id, value
FROM readings
WHERE sensor_id = ?
ORDER BY observed_at DESC
LIMIT 1
""", (1,)).fetchall()
for step in plan:
print(tuple(step))索引應依常用查詢建立，不是越多越好；每個索引都會占空間，也增加新增與更新資料的成本。
十六、課堂挑戰
挑戰A｜基礎：查出所有
quality='unknown' 的紀錄，依時間由新到舊排列。挑戰B｜進階：列出每座場域的感測器數量，沒有設備的場域也必須顯示為0。
挑戰C｜USR場域：和養殖者共同定義「設備待確認、數值待確認、正常追蹤」三類，不記錄個資，並讓規則附上來源、適用物種與版本日期。
十七、用AI協助，但要自己驗證
請擔任SQL學習助教。資料表為sites、sensors、readings。
請先用一句話說明查詢邏輯，再提供SQLite相容SQL。
所有外部資料值必須參數化；不得虛構欄位。
我的問題是：找出每座場域最新一筆有效水溫，
沒有有效資料的場域也要顯示。
最後請提供兩筆可驗證LEFT JOIN行為的最小測試資料。收到AI產生的SQL後，至少檢查欄名、JOIN條件、NULL處理、重複列、參數化及測試資料；能執行不代表答案一定正確。
沒有留言:
張貼留言