《Python一下：從風土資料到智慧生活》第 27 篇
從SQLite走向MySQL——建立多人共用的智慧養殖資料庫
讓場域設備、學生團隊與管理平台安全共用資料，而不是共用一個資料庫檔案。
學習目標
完成本篇後，你能比較SQLite與MySQL的適用情境，建立UTF-8資料庫及最小權限帳號，使用環境變數與PyMySQL安全連線，以交易完成批次寫入、查詢與更新，並說明遠端部署、備份及個資保護的基本原則。
完成本篇後，你能比較SQLite與MySQL的適用情境，建立UTF-8資料庫及最小權限帳號，使用環境變數與PyMySQL安全連線，以交易完成批次寫入、查詢與更新，並說明遠端部署、備份及個資保護的基本原則。
一、SQLite很好，為何還需要MySQL？
SQLite把整個資料庫放在單一檔案中，適合個人練習、離線程式、小型裝置及單機原型。當多座智慧養殖場域、Django平台與學生團隊需要同時存取，伺服器型資料庫較容易集中管理帳號、連線、權限、交易與備份。
| 比較 | SQLite | MySQL |
|---|---|---|
| 架構 | 程式直接讀寫檔案 | 用戶端連線至資料庫伺服器 |
| 安裝 | Python內建sqlite3 | 需安裝、啟動及維護服務 |
| 多人並行 | 讀取方便，寫入並行有限 | 適合多使用者與網路服務 |
| 帳號權限 | 主要依賴檔案權限 | 可依帳號、主機與資料庫授權 |
| 適合本系列 | 課堂練習、離線採集 | 多人共用平台與正式場域 |
不是資料越多就一定要換MySQL：是否遷移應看同時使用者、寫入頻率、權限、維運能力、可用性及備份需求。若SQLite已能安全滿足需求，不必為了「看起來專業」增加系統複雜度。
二、準備MySQL與Python驅動程式
先由教師或系統管理者安裝受支援版本的MySQL Server。Python端使用PyMySQL：
import subprocess
import sys
subprocess.run(
[sys.executable, "-m", "pip", "install", "PyMySQL"],
check=True,
)在終端機直接執行時，也可輸入 python -m pip install PyMySQL。教室與正式專案應使用虛擬環境，並將套件版本記錄於requirements.txt。
三、由管理者建立資料庫與專用帳號
以下SQL應由有權限的管理者在MySQL工具中執行；請替換密碼，且不要直接開放給整個網際網路。
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)10.0.0.%只是私有網段示例，必須依實際網路縮小允許來源。應另外設計遷移帳號負責CREATE與ALTER；平常執行的應用程式帳號不需要DROP、GRANT等高權限。
四、把連線設定放在環境變數
不要把密碼寫進Python、Colab、GitHub或部落格。不同作業系統的設定方式不同，程式只負責讀取：
import os
DB_CONFIG = {
"host": os.getenv("SHUIJING_DB_HOST", "127.0.0.1"),
"port": int(os.getenv("SHUIJING_DB_PORT", "3306")),
"user": os.getenv("SHUIJING_DB_USER", "shuijing_app"),
"password": os.getenv("SHUIJING_DB_PASSWORD", ""),
"database": os.getenv("SHUIJING_DB_NAME", "shuijing_demo"),
}
if not DB_CONFIG["password"]:
raise RuntimeError("尚未設定SHUIJING_DB_PASSWORD")Google Colab提醒：不要在公開Notebook中顯示密碼。可使用Colab的Secrets功能讀取密鑰；資料庫也不應直接暴露3306連接埠給所有來源。教學可使用校內測試網段、VPN、SSH通道或受控雲端環境。
五、建立安全連線函式
import pymysql
def connect_database(config):
return pymysql.connect(
host=config["host"],
port=config["port"],
user=config["user"],
password=config["password"],
database=config["database"],
charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor,
autocommit=False,
connect_timeout=10,
read_timeout=10,
write_timeout=10,
)
connection = connect_database(DB_CONFIG)
with connection.cursor() as cursor:
cursor.execute("SELECT VERSION() AS version")
print(cursor.fetchone())連線逾時避免程式無限等待；DictCursor讓查詢結果以欄名存取。正式遠端連線應依部署環境驗證TLS憑證，不應只為方便而關閉驗證。
六、建立三張InnoDB關聯表
SCHEMA_SQL = [
"""
CREATE TABLE IF NOT EXISTS sites (
site_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
site_code VARCHAR(30) NOT NULL UNIQUE,
display_name VARCHAR(100) NOT NULL,
active BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""",
"""
CREATE TABLE IF NOT EXISTS sensors (
sensor_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
sensor_code VARCHAR(50) NOT NULL UNIQUE,
site_id BIGINT UNSIGNED NOT NULL,
kind VARCHAR(30) NOT NULL,
unit VARCHAR(20) NOT NULL,
CONSTRAINT fk_sensors_site
FOREIGN KEY (site_id) REFERENCES sites(site_id)
ON UPDATE CASCADE ON DELETE RESTRICT
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""",
"""
CREATE TABLE IF NOT EXISTS readings (
reading_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
sensor_id BIGINT UNSIGNED NOT NULL,
observed_at DATETIME(6) NOT NULL,
value DOUBLE NULL,
quality ENUM('valid','unknown','invalid') NOT NULL,
note VARCHAR(255) NOT NULL DEFAULT '',
received_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uq_sensor_time (sensor_id, observed_at),
KEY idx_readings_time (observed_at),
CONSTRAINT fk_readings_sensor
FOREIGN KEY (sensor_id) REFERENCES sensors(sensor_id)
ON DELETE RESTRICT
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""",
]
try:
with connection.cursor() as cursor:
for statement in SCHEMA_SQL:
cursor.execute(statement)
connection.commit()
except Exception:
connection.rollback()
raise使用InnoDB才能可靠支援交易與外鍵。時間欄位的時區策略要先約定：常見做法是以UTC寫入 DATETIME，應用程式顯示時再轉成臺灣時間，避免夏令時間或跨區部署造成歧義。
七、SQLite與MySQL最容易混淆的差異
| 用途 | SQLite | PyMySQL／MySQL |
|---|---|---|
| 參數佔位符 | ? | %s |
| 自動編號 | INTEGER PRIMARY KEY | AUTO_INCREMENT |
| 真假值 | 通常以0與1儲存 | BOOLEAN實際為TINYINT(1) |
| 新增或忽略 | ON CONFLICT DO NOTHING | INSERT IGNORE或ON DUPLICATE KEY UPDATE |
| 列資料 | sqlite3.Row | DictCursor |
八、用參數化SQL新增場域
def add_site(connection, site_code, display_name):
sql = """
INSERT INTO sites (site_code, display_name)
VALUES (%s, %s)
"""
try:
with connection.cursor() as cursor:
cursor.execute(sql, (site_code, display_name))
connection.commit()
except pymysql.MySQLError:
connection.rollback()
raise
add_site(connection, "SITE-01", "示範池A")即使PyMySQL的佔位符長得像字串格式化，也不能自己使用 % 或f-string拼接SQL；參數必須交給 execute() 的第二個參數處理。
九、重複執行也安全的Upsert
def upsert_site(connection, site_code, display_name):
sql = """
INSERT INTO sites (site_code, display_name)
VALUES (%s, %s) AS new
ON DUPLICATE KEY UPDATE
display_name = new.display_name
"""
try:
with connection.cursor() as cursor:
cursor.execute(sql, (site_code, display_name))
connection.commit()
except pymysql.MySQLError:
connection.rollback()
raise
upsert_site(connection, "SITE-01", "示範池A")Upsert可讓同步工作重跑而不產生重複場域，但更新哪些欄位仍需明確規劃，避免新資料不小心覆蓋人工校正內容。
十、交易：場域與感測器一起成功
def create_site_with_sensor(
connection, site_code, display_name,
sensor_code, kind, unit
):
try:
with connection.cursor() as cursor:
cursor.execute(
"""
INSERT INTO sites (site_code, display_name)
VALUES (%s, %s)
""",
(site_code, display_name),
)
site_id = cursor.lastrowid
cursor.execute(
"""
INSERT INTO sensors
(sensor_code, site_id, kind, unit)
VALUES (%s, %s, %s, %s)
""",
(sensor_code, site_id, kind, unit),
)
connection.commit()
except pymysql.MySQLError:
connection.rollback()
raise任何一步失敗都回滾，避免只留下沒有設備的半套資料。不要在尚未完成整個工作單元前提早commit。
十一、批次寫入感測資料
def insert_readings(connection, sensor_id, items):
sql = """
INSERT INTO readings
(sensor_id, observed_at, value, quality, note)
VALUES (%s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
value = VALUES(value),
quality = VALUES(quality),
note = VALUES(note)
"""
rows = [
(
sensor_id,
item["observed_at"],
item.get("value"),
item["quality"],
item.get("note", ""),
)
for item in items
]
try:
with connection.cursor() as cursor:
cursor.executemany(sql, rows)
connection.commit()
except pymysql.MySQLError:
connection.rollback()
raise先驗證再入庫：程式仍應檢查時間格式、資料型別、quality允許值與合理範圍。超出範圍的原始數值可標成invalid保留，以利追查；設備離線則用NULL加unknown，不要偽裝成0。
十二、JOIN查出最新有效水溫
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()
raiseFOR 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()正式Log不要輸出密碼、完整連線字串、Token、住戶姓名或精確座標。網站服務通常使用連線池；每次請求都新建連線會浪費資源，下篇SQLAlchemy會再處理。
十五、從SQLite搬到MySQL的正確流程
- 盤點SQLite結構、型別、外鍵、索引、NULL及重複資料。
- 先在MySQL測試環境建立結構，不直接動正式系統。
- 以Python分批讀取SQLite、驗證、轉換時間與寫入MySQL。
- 比對每表筆數、抽樣內容、外鍵及彙總結果。
- 安排短暫停止寫入或雙寫切換，保留可回復方案。
- 確認應用程式穩定後，再依保存政策封存舊資料。
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表名不能使用一般參數佔位符，所以只能從程式內的允許清單選取。筆數相等只是第一關，還要比對內容、關聯與統計結果。
十六、備份不是複製資料夾
MySQL運作中不能只複製資料目錄。應使用資料庫提供的邏輯備份或經驗證的實體備份機制，安排自動化、異地保存、加密、保存期限及定期還原演練。
BACKUP_CHECKLIST = [
"備份範圍包含結構、資料、觸發器與必要帳號設定",
"備份檔加密，存取權限與正式資料一致或更嚴格",
"至少保存一份於不同故障範圍",
"記錄備份時間、版本、雜湊值與執行結果",
"定期在隔離環境實際還原並驗證查詢",
]
for number, item in enumerate(BACKUP_CHECKLIST, start=1):
print(number, item)十七、水井村USR的資料治理界線
| 做法 | 原因 |
|---|---|
| 教材使用匿名場域代碼 | 避免揭露養殖戶身分與精確位置 |
| 設備只寫入指定表格 | 降低憑證外洩造成的損害 |
| 管理者、教師、學生分開帳號 | 便於停權、稽核與責任釐清 |
| 平台不直接回傳所有原始資料 | 依角色與任務提供最少資料 |
| 保留品質標記與規則版本 | 避免把設備異常誤認為養殖異常 |
十八、課堂挑戰
挑戰A｜基礎：建立一個只能SELECT指定資料表的唯讀帳號，說明它和應用程式讀寫帳號的差別。
挑戰B｜進階：設計Python匯入程式，將1000筆資料分批寫入，每批失敗時只回滾該批，並記錄可重跑的錯誤清單。
挑戰C｜USR場域：畫出「感測器→閘道器→API→MySQL→儀表板」的資料流，為每一段標示傳輸加密、身分驗證、最小權限與斷線補傳策略。
十九、用AI協助檢查遷移設計
請擔任MySQL資料庫助教，檢查SQLite移轉到MySQL的設計。
資料表為sites、sensors、readings，使用Python與PyMySQL。
請逐項檢查：資料型別、utf8mb4、時區、NULL、外鍵、
唯一鍵、索引、參數化SQL、交易、最小權限、TLS、備份與回復。
不要要求我貼出密碼或真實個資。
請將「一定要改」「建議改進」「需要場域確認」分成三類，
並為每一項提供可驗證方法。
沒有留言:
張貼留言