from sqlalchemy.orm import Session
with Session(engine) as session:
with session.begin():
site = Site(
site_code="SITE-01",
display_name="示範池A",
)
session.add(site)
print("場域新增完成")
from sqlalchemy import select
statement = (
select(Site)
.where(Site.active.is_(True))
.order_by(Site.site_code)
)
with Session(engine) as session:
sites = session.scalars(statement).all()
for site in sites:
print(site.site_code, site.display_name)
from sqlalchemy.orm import selectinload
statement = (
select(Site)
.options(selectinload(Site.sensors))
.order_by(Site.site_code)
)
with Session(engine) as session:
sites = session.scalars(statement).all()
for site in sites:
print(site.site_code, len(site.sensors))
from datetime import datetime
def build_reading(observed_at, value, quality, note=""):
if quality not in {"valid", "unknown", "invalid"}:
raise ValueError("不支援的資料品質")
if not isinstance(observed_at, datetime):
raise TypeError("observed_at必須是datetime")
if quality == "unknown":
value = None
elif value is None:
raise ValueError("valid或invalid必須保留原始值")
else:
value = float(value)
return Reading(
observed_at=observed_at,
value=value,
quality=quality,
note=str(note)[:255],
)
def register_sensor(session, site_code, sensor_code, kind, unit):
repository = SiteRepository(session)
site = repository.get_by_code(site_code)
if site is None:
raise ValueError("場域不存在")
duplicate = session.scalar(
select(Sensor).where(Sensor.sensor_code == sensor_code)
)
if duplicate is not None:
raise ValueError("感測器代碼已存在")
sensor = Sensor(
sensor_code=sensor_code,
kind=kind,
unit=unit,
site=site,
)
session.add(sensor)
return sensor
with Session(engine) as session:
with session.begin():
register_sensor(
session, "SITE-03", "TEMP-03",
"temperature", "°C",
)
十七、用SQLite快速測試相同Model
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
test_engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(test_engine)
with Session(test_engine) as session:
with session.begin():
repository = SiteRepository(session)
repository.add("TEST-01", "測試場域")
with Session(test_engine) as session:
saved = session.scalar(
select(Site).where(Site.site_code == "TEST-01")
)
assert saved is not None
assert saved.display_name == "測試場域"
print("測試通過")
ADMIN_SQL = """
CREATE DATABASE IF NOT EXISTS shuijing_demo
CHARACTER SET utf8mb4
COLLATE utf8mb4_0900_ai_ci;
CREATE USER IF NOT EXISTS
'shuijing_app'@'10.0.0.%'
IDENTIFIED BY '請換成長且唯一的密碼';
GRANT SELECT, INSERT, UPDATE, DELETE
ON shuijing_demo.*
TO 'shuijing_app'@'10.0.0.%';
FLUSH PRIVILEGES;
"""
print(ADMIN_SQL)
def latest_temperature(connection, site_code):
sql = """
SELECT s.site_code,
s.display_name,
e.sensor_code,
r.observed_at,
r.value,
e.unit
FROM sites AS s
JOIN sensors AS e ON e.site_id = s.site_id
JOIN readings AS r ON r.sensor_id = e.sensor_id
WHERE s.site_code = %s
AND e.kind = 'temperature'
AND r.quality = 'valid'
ORDER BY r.observed_at DESC, r.reading_id DESC
LIMIT 1
"""
with connection.cursor() as cursor:
cursor.execute(sql, (site_code,))
return cursor.fetchone()
print(latest_temperature(connection, "SITE-01"))
十三、鎖定資料後更新：避免互相覆蓋
def deactivate_site(connection, site_code):
try:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT site_id, active
FROM sites
WHERE site_code = %s
FOR UPDATE
""",
(site_code,),
)
site = cursor.fetchone()
if site is None:
raise ValueError("找不到場域")
cursor.execute(
"UPDATE sites SET active = FALSE WHERE site_id = %s",
(site["site_id"],),
)
connection.commit()
except Exception:
connection.rollback()
raise
FOR UPDATE需置於交易中，適合「先讀再改」且不能被他人同時改動的流程。鎖定範圍與交易時間應盡量縮小，以免造成等待或死結。
十四、連線生命週期與錯誤紀錄
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("shuijing-db")
def run_job(config):
connection = None
try:
connection = connect_database(config)
with connection.cursor() as cursor:
cursor.execute("SELECT 1 AS healthy")
logger.info("database health=%s", cursor.fetchone()["healthy"])
except pymysql.MySQLError as error:
logger.error("資料庫作業失敗：%s", type(error).__name__)
raise
finally:
if connection is not None:
connection.close()
def compare_counts(sqlite_connection, mysql_connection, table):
allowed = {"sites", "sensors", "readings"}
if table not in allowed:
raise ValueError("不允許的資料表")
sqlite_count = sqlite_connection.execute(
f"SELECT COUNT(*) FROM {table}"
).fetchone()[0]
with mysql_connection.cursor() as cursor:
cursor.execute(f"SELECT COUNT(*) AS total FROM {table}")
mysql_count = cursor.fetchone()["total"]
return sqlite_count, mysql_count
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 "尚無資料")
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"])
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))
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))
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))
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))
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 "找不到")
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))
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))
import csv
from pathlib import Path
path = Path("巡查資料_字典版.csv")
with path.open("r", encoding="utf-8-sig", newline="") as file:
reader = csv.DictReader(file)
print("標題列：", reader.fieldnames)
for record in reader:
temperature = float(record["水溫"])
ph = float(record["pH"])
print(record["場域"], temperature, ph)
六、先檢查必要欄位
def validate_header(fieldnames):
required = {"日期", "場域", "水溫", "pH", "狀態"}
actual = set(fieldnames or [])
missing = required - actual
if missing:
names = "、".join(sorted(missing))
raise ValueError(f"CSV缺少必要欄位：{names}")
validate_header(["日期", "場域", "水溫", "pH", "狀態"])
print("標題列檢查完成")
欄位檢查應在逐筆轉換之前完成，否則資料跑到一半才發現整份格式不相容。
七、驗證每一列並隔離錯誤
from datetime import date
def parse_record(record):
patrol_date = date.fromisoformat(record["日期"].strip())
site = record["場域"].strip()
if not site:
raise ValueError("場域不可空白")
temperature = float(record["水溫"])
ph = float(record["pH"])
if not 0.0 <= ph <= 14.0:
raise ValueError("pH超出理論範圍")
return {
"日期": patrol_date.isoformat(),
"場域": site,
"水溫": temperature,
"pH": ph,
"狀態": record["狀態"].strip(),
}
demo_records = [
{"日期": "2026-08-31", "場域": f"示範池{letter}",
"水溫": 26.0 + index * 0.4,
"pH": 7.0 + index * 0.1,
"狀態": "教學資料"}
for index, letter in enumerate("ABCDEFG")
]
for record in demo_records:
print(record)
十、計算摘要，不把錯誤值混進去
def summarize(records):
if not records:
return {"筆數": 0, "平均水溫": None, "平均pH": None}
temperature_average = sum(
record["水溫"] for record in records
) / len(records)
ph_average = sum(
record["pH"] for record in records
) / len(records)
return {
"筆數": len(records),
"平均水溫": round(temperature_average, 2),
"平均pH": round(ph_average, 2),
}
print(summarize(demo_records))
def safe_csv_text(value):
text = str(value)
if text.startswith(("=", "+", "-", "@")):
return "'" + text
return text
examples = ["正常", '=HYPERLINK("bad")', "@command"]
for value in examples:
print(safe_csv_text(value))
