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

Prerequisites and exercise order

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

• The example table belongs to an external business database. Do not write directly to Mingcedar tag_latest or runtime configuration tables.
• 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.
• 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.

1. Run the SQL exercise without additional packages

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.

1. Select the practice project on Code and create sql_basics under scripts/.
2. Copy the complete code, save, select sql_basics.py, and click Run Test.
3. Inspect the output, then run again. It should still contain one row because each run creates a new in-memory database.

# 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({})

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

2. Prepare the external business table

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.

1. Obtain the business database host, port, database name, script account, and CA file from its administrator. Confirm the target is the external business database.
2. Execute the CREATE TABLE statement once in that database’s SQL client, not in the Python editor.
3. Verify that integration_measurements exists and the script account can SELECT, INSERT, and UPDATE it.
4. Choose a record_id for this exercise and use it consistently for insert, query, and update. Do not store the literal <record-id> placeholder.

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

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

3. Configure the address, certificate, and password

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.

1. Select the project, expand Secrets, enter DATABASE_PASSWORD as the name and the database password as the value, then click Create Secret.
2. Confirm DATABASE_PASSWORD appears in the list. Its value is not shown again; do not put it in ordinary environment variables or logs.
3. 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.
4. On Code, create business_config under common/ and copy the template below. Keep the configuration for your database branch.
5. 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.
6. Save the file. The code reads the password from DATABASE_PASSWORD; do not hardcode it.

# 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,
}

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

4A. Save the PostgreSQL insert function

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.

1. Run the dependency probe and verify that psycopg imports successfully.
2. Create postgres_store under common/, copy the complete function below, and save.
3. This file only defines database functions; later entrypoints import and call them.

# 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]}

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

4A continued: query and update functions

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

# 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}

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

4B. Save the MySQL insert function

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.

1. Run the dependency probe and verify that pymysql imports successfully.
2. Create mysql_store under common/, copy the complete function below, and save.
3. This file only defines database functions; later entrypoints import and call them.

# 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()

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

4B continued: query and update functions

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.

# 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()

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

5. Run a read-only query first

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

1. Keep postgres_store and POSTGRES_CONFIG for PostgreSQL; change both to mysql_store and MYSQL_CONFIG for MySQL.
2. Inspect rows. Each page contains at most 100 records ordered by record_id.
3. 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.

# 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": ""})

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

6. Insert one business record

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.

1. Create store_measurement under scripts/, copy the complete code, and select the same database branch as the query script.
2. Replace <record-id> and tag_ref at the bottom; keep value=12.5.
3. Select store_measurement.py and execute Run Test once.
4. Run read_measurements.py again and find that record_id with the expected tag_ref and 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})

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.

7. Update the record and query again

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

1. Adjust both imports for your database.
2. Replace <record-id> with the ID just inserted, save, and run once.
3. Run the read-only query again: the record should have value=13.0 without increasing the row count.
4. Try an ID that does not exist. Expect Record does not exist, and do not convert that error into success.

# 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})

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.

Database connection and execution issues checks

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

• Connection error: check host, port, DNS, certificates, and account from the actual Script service/container network. Keep TLS verification enabled.
• Type error: verify SQL column types and Python values. Use explicit timezones and parameter binding.
• 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.
• Verify duplicate business requests do not create duplicate rows, failures remain visible, and connections close.
• 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.