所属产品: Mingcedar CONTROL

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

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

首次发布 2026-10-25T12:48:42.557201446Z 最近更新 2026-10-25T12:48:42.557200685Z

<div class="manual-doc-article manual-doc-article-section"><h2 id="python-sql">开发教程: Python 与 SQL：查询、写入及公共数据访问方法</h2>

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

<h3 id="python-sql-item-1">准备与练习顺序</h3>

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

<ul>

<li>示例表属于外部业务数据库，不是 Mingcedar 内部运行数据库；不要直接改写 tag_latest 或运行配置表。</li>

<li>Run Test 会执行脚本中的普通 Python 调用；SQL、HTTP 和邮件操作可能实际生效，请先连接独立的练习数据库或接口。</li>

<li>连接信息写入 common/business_config.py；密码通过当前项目的 Secret DATABASE_PASSWORD 读取。这里使用的是项目显式配置的环境，不依赖宿主机环境变量继承。</li>

</ul>

</article>

<h3 id="python-sql-item-2">1. 运行无需安装的 SQL 练习</h3>

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

<ol>

<li>在代码页选择练习项目，在 scripts/ 新建 sql_basics。</li>

<li>复制完整代码并保存；选中 sql_basics.py，点击 Run Test。</li>

<li>观察输出后再运行一次，仍应只有一行。本示例每次建立新的内存数据库。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">PythonPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: Python">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># 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({})

</code></pre></div>

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

</article>

<h3 id="python-sql-item-3">2. 准备外部业务表</h3>

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

<ol>

<li>由业务数据库管理员提供数据库地址、端口、库名、脚本账号和 CA 文件。确认这是外部业务库。</li>

<li>在该库的 SQL 管理工具中执行下方建表语句一次。不要粘贴到 Python 编辑器。</li>

<li>确认 integration_measurements 表存在，脚本账号可以 SELECT、INSERT、UPDATE 此表。</li>

<li>确定本次练习 record_id，后续插入、查询、更新使用同一个标识；不要把 <record-id> 原样写入库。</li>

</ol>

<div class="manual-code" data-code-language="sql"><div class="manual-code-toolbar">SQLSQL<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: SQL">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code>CREATE TABLE integration_measurements (

record_id VARCHAR(128) PRIMARY KEY,

tag_ref VARCHAR(512) NOT NULL,

value DOUBLE PRECISION NOT NULL

);</code></pre></div>

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

</article>

<h3 id="python-sql-item-6">3. 配置数据库地址、证书与密码</h3>

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

<ol>

<li>选择当前项目，展开 Secrets，在名称填 DATABASE_PASSWORD，在值中填写数据库管理员提供的密码，点击“创建 Secret（Create Secret）”。</li>

<li>确认 Secret 列表出现 DATABASE_PASSWORD。值提交后不回显；不要把密码放入普通环境变量或日志。</li>

<li>点击“保存并应用配置”，让新的运行版本使用该配置修订。仅创建或替换 Secret 不会改变已发布的旧版本。</li>

<li>回到代码页，在 common/ 创建 business_config，复制下方模板；只保留所选数据库对应配置。</li>

<li>替换 host、库名、user 和 CA 路径。PostgreSQL 通常使用 5432，MySQL 通常使用 3306，以实际部署为准。容器中的 CA 路径必须在容器内可读。</li>

<li>保存该文件。密码由 DATABASE_PASSWORD 获取，代码中无需再填写密码。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">common/business_config.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: common/business_config.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># 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,

}

</code></pre></div>

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

<div class="doc-step-troubleshooting">如果结果不一致<ul>

<li>KeyError: DATABASE_PASSWORD：回到所选项目确认 Secret 名称拼写、保存结果和当前使用的配置版本。</li>

<li>证书错误：核对 host 与证书主机名、CA 文件路径及读取权限；保持证书校验开启。</li>

</ul></div>

</article>

<h3 id="python-sql-item-4">4A. 保存 PostgreSQL 写入方法</h3>

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

<ol>

<li>先运行 Python 教程的依赖检查，确认 psycopg 可导入。</li>

<li>在 common/ 新建 postgres_store，复制下方完整方法，点击保存。</li>

<li>该文件只保存数据库方法，不需要单独运行；后续脚本通过 import 调用。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">common/postgres_store.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: common/postgres_store.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># 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]}

</code></pre></div>

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

</article>

<h3 id="sql-postgres-query-update">4A（续）. 查询与更新方法</h3>

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

def list_measurements(config, after=""):

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):

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),

)

</code></pre></div>

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

</article>

<h3 id="python-sql-item-5">4B. 保存 MySQL 写入方法</h3>

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

<ol>

<li>先运行 Python 教程的依赖检查，确认 pymysql 可导入。</li>

<li>在 common/ 新建 mysql_store，复制下方完整方法，点击保存。</li>

<li>该文件只保存数据库方法，不需要单独运行；后续脚本通过 import 调用。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">common/mysql_store.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: common/mysql_store.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># 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:

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(

(record_id,),

)

stored = cursor.fetchone()

if stored != (tag_ref, value):

connection.commit()

except Exception:

connection.rollback()

raise

finally:

connection.close()

</code></pre></div>

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

</article>

<h3 id="sql-mysql-query-update">4B（续）. 查询与更新方法</h3>

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

read_timeout=5, write_timeout=5, autocommit=False)

try:

cursor.execute(

"SELECT record_id, tag_ref, value FROM integration_measurements "

)

return cursor.fetchall()

finally:

connection.close()

def update_measurement(config, record_id, value):

try:

cursor.execute(

"SELECT record_id FROM integration_measurements "

)

raise ValueError("Record does not exist")

cursor.execute(

(value, record_id),

)

connection.commit()

except Exception:

connection.rollback()

raise

finally:

connection.close()

</code></pre></div>

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

</article>

<h3 id="python-sql-item-9">5. 先运行只读查询</h3>

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

<ol>

<li>PostgreSQL 保持 postgres_store 和 POSTGRES_CONFIG；MySQL 同时改成 mysql_store 和 MYSQL_CONFIG。</li>

<li>检查返回 rows。每页最多 100 行，按 record_id 排序。</li>

<li>读取下一页时，把入口中的 after 从空字符串改为上一页返回的 next_after，再运行；空页表示结束。此顺序不表示采集时间顺序。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">scripts/read_measurements.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: scripts/read_measurements.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># 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}

main({"after": ""})

</code></pre></div>

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

</article>

<h3 id="python-sql-item-7">6. 插入一条业务记录</h3>

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

<ol>

<li>在 scripts/ 创建 store_measurement，复制下方完整代码，并保持数据库分支与查询脚本一致。</li>

<li>替换入口底部的 <record-id> 和 tag_ref；value 保持 12.5。</li>

<li>选中 store_measurement.py，通过 Run Test 执行一次。</li>

<li>重新运行 read_measurements.py，找到相同 record_id，核对 tag_ref 和 value=12.5。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">scripts/store_measurement.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: scripts/store_measurement.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># scripts/store_measurement.py

# 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"])

)

"tag_ref": "hq://io/<io-domain-id>/<device-uid>/<tag-uid>", "value": 12.5})

</code></pre></div>

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

</article>

<h3 id="python-sql-item-8">7. 更新记录并重新查询</h3>

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

<ol>

<li>按数据库分支调整两条 import。</li>

<li>把 <record-id> 改为刚写入的标识，保存并运行一次。</li>

<li>重新运行只读查询，核对该记录 value=13.0，记录总数未增加。</li>

<li>再使用一个不存在的标识验证更新，预期 Record does not exist；不要把该错误转换成成功。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">scripts/update_measurement.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: scripts/update_measurement.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># scripts/update_measurement.py

# 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

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

</code></pre></div>

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

</article>

<h3 id="python-sql-item-10">数据库连接与运行问题</h3>

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

<ul>

<li>连接失败：从实际 Script 服务或容器所在网络检查地址、端口、DNS、证书及账号；不要关闭 TLS 校验来绕过错误。</li>

<li>SQL 类型错误：核对列类型和 Python 值；日期使用明确时区，不把字符串拼进 SQL。</li>

<li>执行或提交超时：连接丢失时事务可能已经提交；按 record_id 只读查询业务库后再决定下一步，不盲目重试。</li>

<li>确认重复业务请求不会产生重复记录；确认异常会保留失败状态，连接会关闭。</li>

<li>事务只覆盖本数据库。一次 PLC 写入与一次 SQL 提交不构成跨系统原子事务，需用业务状态和核对流程处理。</li>

</ul>

</article>

</div>