Product: Mingcedar CONTROL

Python and SQL: queries, writes, and shared data access | Mingcedar

Complete shared modules and offline dependencies, then practice SQLite, business-database access, inserts, queries, and updates. Choose either PostgreSQL or MySQL.

Published 2026-10-25T12:48:42.637740576Z Updated 2026-10-25T12:48:42.637739634Z

<div class="manual-doc-article manual-doc-article-section"><h2 id="python-sql">Developer tutorial: Python and SQL: queries, writes, and shared data access</h2>

Practice with the standard library, then connect to an external business database using psycopg or PyMySQL with parameterized queries, transactions, and duplicate-request reconciliation.

<h3 id="python-sql-item-1">Prerequisites and exercise order</h3>

Complete shared modules and offline dependencies, then practice SQLite, business-database access, inserts, queries, and updates. Choose either PostgreSQL or MySQL.

<ul>

<li>The example table belongs to an external business database. Do not write directly to Mingcedar tag_latest or runtime configuration tables.</li>

<li>Run Test executes ordinary Python calls in the script. SQL, HTTP, and mail operations may take effect; connect to a separate practice database or endpoint first.</li>

<li>Connection settings live in common/business_config.py. The password comes from this project’s DATABASE_PASSWORD Secret, explicitly configured for the project rather than inherited from the host.</li>

</ul>

</article>

<h3 id="python-sql-item-2">1. Run the SQL exercise without additional packages</h3>

Save as scripts/sql_basics.py and click Run Test. It creates and modifies an in-memory table, returning [('measurement-001', 13.0)]. Data does not persist after execution.

<ol>

<li>Select the practice project on Code and create sql_basics under scripts/.</li>

<li>Copy the complete code, save, select sql_basics.py, and click Run Test.</li>

<li>Inspect the output, then run again. It should still contain one row because each run creates a new in-memory database.</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">PythonPython<button type="button" data-copy-code data-copied="Copied" data-failed="Select and copy the code" aria-label="Copy code: Python">Copy code</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>

Expected result Console shows [('measurement-001', 13.0)]: INSERT stores 12.5, UPDATE changes it to 13.0, and SELECT reads it back.

</article>

<h3 id="python-sql-item-3">2. Prepare the external business table</h3>

Have the maintainer create the table in a dedicated business database and grant the script account access. Reuse record_id for the same record; allocate a new ID for a new record.

<ol>

<li>Obtain the business database host, port, database name, script account, and CA file from its administrator. Confirm the target is the external business database.</li>

<li>Execute the CREATE TABLE statement once in that database’s SQL client, not in the Python editor.</li>

<li>Verify that integration_measurements exists and the script account can SELECT, INSERT, and UPDATE it.</li>

<li>Choose a record_id for this exercise and use it consistently for insert, query, and update. Do not store the literal <record-id> placeholder.</li>

</ol>

<div class="manual-code" data-code-language="sql"><div class="manual-code-toolbar">SQLSQL<button type="button" data-copy-code data-copied="Copied" data-failed="Select and copy the code" aria-label="Copy code: SQL">Copy code</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>

Expected result The business table exists with the three columns shown. No measurement has been written yet.

</article>

<h3 id="python-sql-item-6">3. Configure the address, certificate, and password</h3>

Store the database password in Secrets under Settings → Python → Python Packages → Advanced Project Settings. The configuration below reads that explicitly configured Secret through os.environ.

<ol>

<li>Select the project, expand Secrets, enter DATABASE_PASSWORD as the name and the database password as the value, then click Create Secret.</li>

<li>Confirm DATABASE_PASSWORD appears in the list. Its value is not shown again; do not put it in ordinary environment variables or logs.</li>

<li>Click Save and Apply Configuration so a new runtime version uses this configuration revision. Creating or rotating a Secret does not change existing published versions.</li>

<li>On Code, create business_config under common/ and copy the template below. Keep the configuration for your database branch.</li>

<li>Replace host, database name, user, and CA path. Typical ports are 5432 for PostgreSQL and 3306 for MySQL; use your deployed values. A container must be able to read its CA path.</li>

<li>Save the file. The code reads the password from DATABASE_PASSWORD; do not hardcode it.</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="Copied" data-failed="Select and copy the code" aria-label="Copy code: common/business_config.py">Copy code</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>

Expected result Confirm the Secret, replaced placeholders, and applied configuration. Verify access with a read-only query before inserting.

<div class="doc-step-troubleshooting">If the result differs<ul>

<li>KeyError: DATABASE_PASSWORD: check the selected project, exact Secret name, save result, and configuration version in use.</li>

<li>Certificate error: verify the host against the certificate name, CA path, and file permissions; retain certificate verification.</li>

</ul></div>

</article>

<h3 id="python-sql-item-4">4A. Save the PostgreSQL insert function</h3>

Save as common/postgres_store.py. config supplies host, port, dbname, user, and password, plus deployment-required sslmode/sslrootcert. The connection context commits on normal exit, rolls back on exceptions, and closes the connection. Parameters are passed separately from SQL text.

<ol>

<li>Run the dependency probe and verify that psycopg imports successfully.</li>

<li>Create postgres_store under common/, copy the complete function below, and save.</li>

<li>This file only defines database functions; later entrypoints import and call them.</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="Copied" data-failed="Select and copy the code" aria-label="Copy code: common/postgres_store.py">Copy code</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>

Expected result Use the PostgreSQL module names and configuration constants in subsequent entrypoints.

</article>

<h3 id="sql-postgres-query-update">4A continued: query and update functions</h3>

For PostgreSQL, append this code after save_measurement in common/postgres_store.py. Keep import psycopg at the top.

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>

Expected result The file now defines save_measurement, list_measurements, and update_measurement.

</article>

<h3 id="python-sql-item-5">4B. Save the MySQL insert function</h3>

Save as common/mysql_store.py. config supplies host, port, database, user, password, charset="utf8mb4", and required TLS settings. The function explicitly commits or rolls back and closes the connection. It uses the default tuple cursor.

<ol>

<li>Run the dependency probe and verify that pymysql imports successfully.</li>

<li>Create mysql_store under common/, copy the complete function below, and save.</li>

<li>This file only defines database functions; later entrypoints import and call them.</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="Copied" data-failed="Select and copy the code" aria-label="Copy code: common/mysql_store.py">Copy code</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>

Expected result Use the MySQL module names and configuration constants in subsequent entrypoints.

</article>

<h3 id="sql-mysql-query-update">4B continued: query and update functions</h3>

For MySQL/MariaDB, append this code after save_measurement in common/mysql_store.py. Keep import pymysql. The query always closes its connection; an update failure rolls back.

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>

Expected result The file now defines save_measurement, list_measurements, and update_measurement.

</article>

<h3 id="python-sql-item-9">5. Run a read-only query first</h3>

Create scripts/read_measurements.py, copy the code, adjust both imports for your database, save, and run. An empty-table result [] also confirms access.

<ol>

<li>Keep postgres_store and POSTGRES_CONFIG for PostgreSQL; change both to mysql_store and MYSQL_CONFIG for MySQL.</li>

<li>Inspect rows. Each page contains at most 100 records ordered by record_id.</li>

<li>For the next page, change the entrypoint’s after from an empty string to the returned next_after and run again. An empty page ends traversal; this order is not acquisition-time order.</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="Copied" data-failed="Select and copy the code" aria-label="Copy code: scripts/read_measurements.py">Copy code</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>

Expected result The result is [] or a list of existing records, without connection, authentication, or table-permission errors.

</article>

<h3 id="python-sql-item-7">6. Insert one business record</h3>

Save as scripts/store_measurement.py; replace the record ID, TagRef, and both database imports. Run Test writes value=12.5 to the configured database. First confirm that the target is the business table used for this exercise. Event calls to main(ctx) pass record_id, tag_ref, and value.

<ol>

<li>Create store_measurement under scripts/, copy the complete code, and select the same database branch as the query script.</li>

<li>Replace <record-id> and tag_ref at the bottom; keep value=12.5.</li>

<li>Select store_measurement.py and execute Run Test once.</li>

<li>Run read_measurements.py again and find that record_id with the expected tag_ref and 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="Copied" data-failed="Select and copy the code" aria-label="Copy code: scripts/store_measurement.py">Copy code</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>

Expected result The query script reads the new record. Repeating the same ID and content does not add a row; reusing the ID with different content raises an error.

</article>

<h3 id="python-sql-item-8">7. Update the record and query again</h3>

Create scripts/update_measurement.py. This example explicitly changes the previous record to 13.0. UPDATE has different semantics from repeating an INSERT.

<ol>

<li>Adjust both imports for your database.</li>

<li>Replace <record-id> with the ID just inserted, save, and run once.</li>

<li>Run the read-only query again: the record should have value=13.0 without increasing the row count.</li>

<li>Try an ID that does not exist. Expect Record does not exist, and do not convert that error into success.</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="Copied" data-failed="Select and copy the code" aria-label="Copy code: scripts/update_measurement.py">Copy code</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>

Expected result The existing row is updated. A missing row raises an error and rolls back. Inserting the same ID with the old value=12.5 now reports a payload conflict.

</article>

<h3 id="python-sql-item-10">Database connection and execution issues checks</h3>

Verify connectivity, authentication, TLS, table permissions, parameters, and transactions before scheduling. Do not replace database exceptions with success results.

<ul>

<li>Connection error: check host, port, DNS, certificates, and account from the actual Script service/container network. Keep TLS verification enabled.</li>

<li>Type error: verify SQL column types and Python values. Use explicit timezones and parameter binding.</li>

<li>Execution or commit timeout: a lost connection can leave a committed transaction. Query by record_id before deciding what to do next; do not retry blindly.</li>

<li>Verify duplicate business requests do not create duplicate rows, failures remain visible, and connections close.</li>

<li>A SQL transaction covers only its database. A PLC write and SQL commit are not a cross-system atomic transaction; reconcile them using business state.</li>

</ul>

</article>

</div>