文档手册

Python 与 SQL：查询、写入及公共数据访问方法

先完成公共模块与离线依赖，再依次练习 SQLite、业务库连接、插入、查询和更新。PostgreSQL 与 MySQL 任选一条分支。

开发教程

开发教程: Python 与 SQL：查询、写入及公共数据访问方法

先用标准库完成 SQL 基础练习，再使用 psycopg 或 PyMySQL 连接外部业务数据库，实现参数化查询、事务和重复请求核对。

准备与练习顺序

先完成公共模块与离线依赖，再依次练习 SQLite、业务库连接、插入、查询和更新。PostgreSQL 与 MySQL 任选一条分支。

示例表属于外部业务数据库，不是 Mingcedar 内部运行数据库；不要直接改写 tag_latest 或运行配置表。
Run Test 会执行脚本中的普通 Python 调用；SQL、HTTP 和邮件操作可能实际生效，请先连接独立的练习数据库或接口。
连接信息写入 common/business_config.py；密码通过当前项目的 Secret DATABASE_PASSWORD 读取。这里使用的是项目显式配置的环境，不依赖宿主机环境变量继承。

1. 运行无需安装的 SQL 练习

保存为 scripts/sql_basics.py，点击 Run Test。示例在内存库建表、插入、更新和查询，输出 [('measurement-001', 13.0)]；执行后数据不保留。

在代码页选择练习项目，在 scripts/ 新建 sql_basics。
复制完整代码并保存；选中 sql_basics.py，点击 Run Test。
观察输出后再运行一次，仍应只有一行。本示例每次建立新的内存数据库。

# scripts/sql_basics.py — invocation-local SQL practice
import sqlite3

def main(ctx):
    connection = sqlite3.connect(":memory:")
    try:
        with connection:
            connection.execute("CREATE TABLE measurements (record_id TEXT PRIMARY KEY, value REAL NOT NULL)")
            connection.execute("INSERT INTO measurements VALUES (?, ?)", ("measurement-001", 12.5))
            connection.execute("UPDATE measurements SET value = ? WHERE record_id = ?", (13.0, "measurement-001"))
        rows = connection.execute("SELECT record_id, value FROM measurements WHERE record_id = ?",
                                  ("measurement-001",)).fetchall()
        print(rows)
        return {"rows": rows}
    finally:
        connection.close()

if __name__ == "__main__":
    main({})

完成标志 控制台为 [('measurement-001', 13.0)]：INSERT 写入 12.5，UPDATE 改为 13.0，SELECT 读回 13.0。

2. 准备外部业务表

由维护人员在专用业务库建表并授予脚本账号所需权限。record_id 标识业务记录：同一记录沿用，新记录重新分配。

由业务数据库管理员提供数据库地址、端口、库名、脚本账号和 CA 文件。确认这是外部业务库。
在该库的 SQL 管理工具中执行下方建表语句一次。不要粘贴到 Python 编辑器。
确认 integration_measurements 表存在，脚本账号可以 SELECT、INSERT、UPDATE 此表。
确定本次练习 record_id，后续插入、查询、更新使用同一个标识；不要把 <record-id> 原样写入库。

CREATE TABLE integration_measurements (
    record_id VARCHAR(128) PRIMARY KEY,
    tag_ref VARCHAR(512) NOT NULL,
    value DOUBLE PRECISION NOT NULL
);

完成标志 业务表存在且三列类型与示例一致。当前步骤尚未写入业务记录。

3. 配置数据库地址、证书与密码

使用 Settings → Python → Python 模块 → 高级项目配置中的 Secrets 保存数据库密码。下方配置通过 os.environ 读取这个明确配置的 Secret；不会自动获得操作系统登录用户的密码。

选择当前项目，展开 Secrets，在名称填 DATABASE_PASSWORD，在值中填写数据库管理员提供的密码，点击“创建 Secret（Create Secret）”。
确认 Secret 列表出现 DATABASE_PASSWORD。值提交后不回显；不要把密码放入普通环境变量或日志。
点击“保存并应用配置”，让新的运行版本使用该配置修订。仅创建或替换 Secret 不会改变已发布的旧版本。
回到代码页，在 common/ 创建 business_config，复制下方模板；只保留所选数据库对应配置。
替换 host、库名、user 和 CA 路径。PostgreSQL 通常使用 5432，MySQL 通常使用 3306，以实际部署为准。容器中的 CA 路径必须在容器内可读。
保存该文件。密码由 DATABASE_PASSWORD 获取，代码中无需再填写密码。

# common/business_config.py — configuration template, no real credentials
import os

POSTGRES_CONFIG = {
    "host": "<database-host>", "port": 5432,
    "dbname": "<database-name>", "user": "<database-user>",
    "password": os.environ["DATABASE_PASSWORD"], "sslmode": "verify-full",
    "sslrootcert": "<ca-file-path>",
}
MYSQL_CONFIG = {
    "host": "<database-host>", "port": 3306,
    "database": "<database-name>", "user": "<database-user>",
    "password": os.environ["DATABASE_PASSWORD"], "charset": "utf8mb4",
    "ssl_ca": "<ca-file-path>", "ssl_verify_cert": True,
    "ssl_verify_identity": True,
}

完成标志 Secret 可见、占位符已替换、配置已应用。先以只读查询验证连接，再插入。

如果结果不一致

KeyError: DATABASE_PASSWORD：回到所选项目确认 Secret 名称拼写、保存结果和当前使用的配置版本。
证书错误：核对 host 与证书主机名、CA 文件路径及读取权限；保持证书校验开启。

4A. 保存 PostgreSQL 写入方法

保存为 common/postgres_store.py。config 使用 host、port、dbname、user、password，并按部署要求提供 sslmode/sslrootcert。with 块正常结束提交事务，发生异常回滚并关闭连接；参数通过独立元组传入 SQL。

先运行 Python 教程的依赖检查，确认 psycopg 可导入。
在 common/ 新建 postgres_store，复制下方完整方法，点击保存。
该文件只保存数据库方法，不需要单独运行；后续脚本通过 import 调用。

# common/postgres_store.py
import psycopg

def save_measurement(config, record_id, tag_ref, value):
    with psycopg.connect(**config, connect_timeout=5,
                         options="-c statement_timeout=5000") as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                "INSERT INTO integration_measurements (record_id, tag_ref, value) "
                "VALUES (%s, %s, %s) ON CONFLICT (record_id) DO NOTHING",
                (record_id, tag_ref, value),
            )
            cursor.execute(
                "SELECT tag_ref, value FROM integration_measurements WHERE record_id = %s",
                (record_id,),
            )
            stored = cursor.fetchone()
            if stored != (tag_ref, value):
                raise ValueError("Record ID already exists with a different payload")
    return {"record_id": record_id, "tag_ref": stored[0], "value": stored[1]}

完成标志 后续入口须使用 PostgreSQL 分支的模块名和配置常量。

4A（续）. 查询与更新方法

选择 PostgreSQL 时执行本节。把代码追加到 common/postgres_store.py 的现有 save_measurement 函数之后，保留文件顶部 import psycopg。

# Append to common/postgres_store.py; psycopg is imported above.
def list_measurements(config, after=""):
    with psycopg.connect(**config, connect_timeout=5,
                         options="-c statement_timeout=5000") as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT record_id, tag_ref, value FROM integration_measurements "
                "WHERE record_id > %s ORDER BY record_id LIMIT 100", (after,)
            )
            return cursor.fetchall()

def update_measurement(config, record_id, value):
    with psycopg.connect(**config, connect_timeout=5,
                         options="-c statement_timeout=5000") as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT record_id FROM integration_measurements "
                "WHERE record_id = %s FOR UPDATE", (record_id,)
            )
            if cursor.fetchone() is None:
                raise ValueError("Record does not exist")
            cursor.execute(
                "UPDATE integration_measurements SET value = %s WHERE record_id = %s",
                (value, record_id),
            )
    return {"record_id": record_id, "value": value}

完成标志 同一文件现在包含 save_measurement、list_measurements、update_measurement 三个函数。

4B. 保存 MySQL 写入方法

保存为 common/mysql_store.py。config 使用 host、port、database、user、password、charset="utf8mb4"，TLS 参数按数据库要求设置。代码显式提交、异常回滚并关闭连接，使用默认元组游标。

先运行 Python 教程的依赖检查，确认 pymysql 可导入。
在 common/ 新建 mysql_store，复制下方完整方法，点击保存。
该文件只保存数据库方法，不需要单独运行；后续脚本通过 import 调用。

# common/mysql_store.py
import pymysql

def save_measurement(config, record_id, tag_ref, value):
    connection = pymysql.connect(**config, connect_timeout=5, read_timeout=5,
                                 write_timeout=5, autocommit=False)
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                "INSERT INTO integration_measurements (record_id, tag_ref, value) "
                "VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE record_id = record_id",
                (record_id, tag_ref, value),
            )
            cursor.execute(
                "SELECT tag_ref, value FROM integration_measurements WHERE record_id = %s",
                (record_id,),
            )
            stored = cursor.fetchone()
            if stored != (tag_ref, value):
                raise ValueError("Record ID already exists with a different payload")
        connection.commit()
        return {"record_id": record_id, "tag_ref": stored[0], "value": stored[1]}
    except Exception:
        connection.rollback()
        raise
    finally:
        connection.close()

完成标志 后续入口须使用 MySQL 分支的模块名和配置常量。

4B（续）. 查询与更新方法

选择 MySQL/MariaDB 时执行本节。把代码追加到 common/mysql_store.py 的现有 save_measurement 函数之后，保留 import pymysql。查询方法始终关闭连接；更新失败回滚。

# Append to common/mysql_store.py; pymysql is imported above.
def list_measurements(config, after=""):
    connection = pymysql.connect(**config, connect_timeout=5,
                                 read_timeout=5, write_timeout=5, autocommit=False)
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT record_id, tag_ref, value FROM integration_measurements "
                "WHERE record_id > %s ORDER BY record_id LIMIT 100", (after,)
            )
            return cursor.fetchall()
    finally:
        connection.close()

def update_measurement(config, record_id, value):
    connection = pymysql.connect(**config, connect_timeout=5,
                                 read_timeout=5, write_timeout=5, autocommit=False)
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT record_id FROM integration_measurements "
                "WHERE record_id = %s FOR UPDATE", (record_id,)
            )
            if cursor.fetchone() is None:
                raise ValueError("Record does not exist")
            cursor.execute(
                "UPDATE integration_measurements SET value = %s WHERE record_id = %s",
                (value, record_id),
            )
        connection.commit()
        return {"record_id": record_id, "value": value}
    except Exception:
        connection.rollback()
        raise
    finally:
        connection.close()

完成标志 同一文件现在包含 save_measurement、list_measurements、update_measurement 三个函数。

5. 先运行只读查询

新建 scripts/read_measurements.py，复制代码并按数据库调整两条 import，保存后运行 Run Test。空表返回 [] 也表示连接成功。

PostgreSQL 保持 postgres_store 和 POSTGRES_CONFIG；MySQL 同时改成 mysql_store 和 MYSQL_CONFIG。
检查返回 rows。每页最多 100 行，按 record_id 排序。
读取下一页时，把入口中的 after 从空字符串改为上一页返回的 next_after，再运行；空页表示结束。此顺序不表示采集时间顺序。

# scripts/read_measurements.py
from ..common.postgres_store import list_measurements
from ..common.business_config import POSTGRES_CONFIG as DATABASE_CONFIG
# For MySQL, change postgres_store to mysql_store and POSTGRES_CONFIG to MYSQL_CONFIG.

def main(ctx):
    rows = list_measurements(DATABASE_CONFIG, ctx.get("after", ""))
    print(rows)
    return {"rows": rows, "next_after": rows[-1][0] if rows else None}

if __name__ == "__main__":
    main({"after": ""})

完成标志 得到 [] 或已有记录列表，且无连接、认证或表权限错误。

6. 插入一条业务记录

保存为 scripts/store_measurement.py，替换业务标识、TagRef 和数据库分支的两条 import。点击 Run Test 会向所配置的数据库写入一条 value=12.5 的记录，请先确认目标是本次练习使用的业务表。事件调用 main(ctx) 时传入 record_id、tag_ref、value。

在 scripts/ 创建 store_measurement，复制下方完整代码，并保持数据库分支与查询脚本一致。
替换入口底部的 <record-id> 和 tag_ref；value 保持 12.5。
选中 store_measurement.py，通过 Run Test 执行一次。
重新运行 read_measurements.py，找到相同 record_id，核对 tag_ref 和 value=12.5。

# scripts/store_measurement.py
from ..common.postgres_store import save_measurement
from ..common.business_config import POSTGRES_CONFIG as DATABASE_CONFIG
# For MySQL, use mysql_store and MYSQL_CONFIG instead.

def main(ctx):
    return save_measurement(
        DATABASE_CONFIG, ctx["record_id"], ctx["tag_ref"], float(ctx["value"])
    )

if __name__ == "__main__":
    main({"record_id": "<record-id>",
          "tag_ref": "hq://io/<io-domain-id>/<device-uid>/<tag-uid>", "value": 12.5})

完成标志 查询脚本能读回新记录。再次插入相同标识及相同内容不会新增一行；相同标识但不同内容会报错。

7. 更新记录并重新查询

新建 scripts/update_measurement.py。本例明确把上一节记录改为 13.0；UPDATE 与重复 INSERT 有不同的业务含义。

按数据库分支调整两条 import。
把 <record-id> 改为刚写入的标识，保存并运行一次。
重新运行只读查询，核对该记录 value=13.0，记录总数未增加。
再使用一个不存在的标识验证更新，预期 Record does not exist；不要把该错误转换成成功。

# scripts/update_measurement.py
from ..common.postgres_store import update_measurement
from ..common.business_config import POSTGRES_CONFIG as DATABASE_CONFIG
# For MySQL, change postgres_store to mysql_store and POSTGRES_CONFIG to MYSQL_CONFIG.

def main(ctx):
    result = update_measurement(DATABASE_CONFIG, ctx["record_id"], float(ctx["value"]))
    print(result)
    return result

if __name__ == "__main__":
    main({"record_id": "<record-id>", "value": 13.0})

完成标志 已有行被更新；不存在的行报错并回滚。修改后的记录再以旧 value=12.5 调用插入方法会被识别为内容冲突。

数据库连接与运行问题

依次验证连通、认证、TLS、表权限、参数化 SQL 和事务，再接入周期任务。不要将数据库异常直接替换成成功结果。

连接失败：从实际 Script 服务或容器所在网络检查地址、端口、DNS、证书及账号；不要关闭 TLS 校验来绕过错误。
SQL 类型错误：核对列类型和 Python 值；日期使用明确时区，不把字符串拼进 SQL。
执行或提交超时：连接丢失时事务可能已经提交；按 record_id 只读查询业务库后再决定下一步，不盲目重试。
确认重复业务请求不会产生重复记录；确认异常会保留失败状态，连接会关闭。
事务只覆盖本数据库。一次 PLC 写入与一次 SQL 提交不构成跨系统原子事务，需用业务状态和核对流程处理。