Chapter 19 · Application Integration, Connectors, Pools, ORMs, and Reliability Patterns
Prepared Statements, Parameter Binding, SQL Injection Defense, and Type Mapping
Keep SQL structure separate from data values, use bound parameters and prepared statements correctly, allowlist dynamic identifiers, and verify how Python values map to MySQL NULL, numeric, binary, temporal, and JSON types.
Learning outcomes
A ServiceHub search endpoint concatenates a customer-entered fragment into SQL. A test value containing a quote breaks the statement; a more malicious-looking string can change the predicate structure. Escaping ad hoc is not a data-access design. The application must keep SQL structure under developer control and send untrusted values through the connector parameter channel.
Use Connector/Python bound parameters and server prepared cursors without quoting parameter markers manually.
Reproduce a safe SQL-injection-shaped failure in a disposable database and repair it with parameter binding.
Explain why table/column/order-direction identifiers cannot be supplied as ordinary bound values and require allowlisting.
Verify Python/MySQL mappings for NULL, integers, Decimal, text, binary, datetime, and JSON documents.
Use deterministic integration tests and server/query-plan evidence to prove correctness rather than inspecting source code alone.
The injection demonstration runs only against servicehub_app_lab with a read-only SELECT. It teaches the parser boundary; it does not attempt exploitation of an external system.
The parser boundary: values are not syntax
When the driver executes WHERE customer_name = %s with a separate parameter tuple, the value does not become another SQL keyword, quote, or comment. Connector/Python converts supported Python values to the protocol representation MySQL expects. Prepared cursors additionally use MySQL prepared-statement protocol and can reuse a preparation when the same SQL is executed repeatedly.
import os, mysql.connectorcnx = mysql.connector.connect( host="127.0.0.1", user="servicehub_app", password=os.environ["MYSQL_PASSWORD"], database="servicehub_app_lab",)cur = cnx.cursor(prepared=True)stmt = b"SELECT customer_id, customer_name FROM customers WHERE customer_name = %s"for value in ("Contoso Facilities", "O'Reilly Field Service"): cur.execute(stmt, (value,)) print(value, cur.fetchall())cur.close(); cnx.close()With prepared cursors, %s or ? parameter markers are not surrounded by quotes. The connector supplies the value separately. Parameterization is primarily a correctness/security boundary; do not promise that preparation always reduces query latency for every workload.
Safe failure: show what string concatenation gets wrong
This deliberately flawed example is kept local and only reads the lab table. The first input demonstrates a syntax break caused by an apostrophe. The second illustrates why concatenation can alter a WHERE clause. The repair uses exactly the same business input with binding.
unsafe_input = "Northwind Field Services' OR '1'='1"unsafe_sql = ( "SELECT customer_id, customer_name FROM customers " f"WHERE customer_name = '{unsafe_input}'")print("DISPOSABLE LAB UNSAFE SQL:", unsafe_sql)cur.execute(unsafe_sql)unsafe_rows = cur.fetchall()print("unsafe row count", len(unsafe_rows)) # wrong result: more than one rowassert len(unsafe_rows) > 1# Correct boundary: SQL structure is fixed, value is separate.cur.execute( "SELECT customer_id, customer_name FROM customers WHERE customer_name = %s", (unsafe_input,),)rows = cur.fetchall()assert rows == []The corrected query searches for a literal customer name containing those characters and returns no match. Do not “repair” concatenation by inventing your own quote-replacement function; use the connector API.
Identifiers and ORDER BY directions are structure: allowlist them
Parameter markers stand for data values, not SQL grammar. A user-selected sort field such as created_at or priority is an identifier; ASC/DESC are keywords. Binding them as values produces invalid or meaningless SQL. Map an external choice to a closed internal set.
SORT_COLUMNS = { "created": "created_at", "priority": "priority", "id": "work_order_id",}SORT_DIRECTIONS = {"asc": "ASC", "desc": "DESC"}def list_orders(cur, sort_key: str, direction: str, status: str): column = SORT_COLUMNS.get(sort_key) order = SORT_DIRECTIONS.get(direction.lower()) if column is None or order is None: raise ValueError("unsupported sort option") sql = f""" SELECT work_order_id, status, priority, created_at FROM work_orders WHERE status = %s ORDER BY {column} {order}, work_order_id {order} LIMIT 50 """ cur.execute(sql, (status,)) return cur.fetchall()The f-string is safe here only because the interpolated fragments come from developer-owned constants after validation. The status remains a bound data value.
Type mapping: test the values your application actually sends
Application reliability also fails when a value is secure but mapped incorrectly. Monetary amounts should use Python Decimal rather than binary floating point when exact decimal behavior matters. SQL NULL maps to Python None. Datetimes require an explicit application time-zone convention; this course uses UTC session state and timezone-naive Python values that represent UTC for the simple lab.
from decimal import Decimalfrom datetime import datetimeimport json, os, mysql.connectorcnx = mysql.connector.connect( host="127.0.0.1", user="servicehub_app", password=os.environ["MYSQL_PASSWORD"], database="servicehub_app_lab")cur = cnx.cursor()cur.execute("SET SESSION time_zone = '+00:00'")payload = {"source":"python","flags":["urgent","field"]}fingerprint = bytes.fromhex("00112233445566778899aabbccddeeff")cur.execute("DELETE FROM work_orders WHERE idempotency_key IN (%s,%s)", ("types-001", "types-002"))cur.execute(""" INSERT INTO work_orders (customer_id,idempotency_key,status,priority,summary,estimated_cost,due_at,metadata,request_fingerprint) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)""", ( 4, "types-001", "open", 2, "Type mapping test", Decimal("19.95"), datetime(2026,8,31,12,30,0), json.dumps(payload), fingerprint))new_id = cur.lastrowidcur.execute(""" INSERT INTO work_orders (customer_id,idempotency_key,status,priority,summary,estimated_cost,due_at,metadata,request_fingerprint) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)""", (4, "types-002", "open", 3, "NULL mapping test", None, None, None, None))cnx.commit()cur.execute(""" SELECT work_order_id, priority, summary, estimated_cost, due_at, metadata, HEX(request_fingerprint) FROM work_orders WHERE work_order_id=%s""", (new_id,))print(cur.fetchone())cur.execute(""" SELECT estimated_cost IS NULL, due_at IS NULL, metadata IS NULL, request_fingerprint IS NULL FROM work_orders WHERE idempotency_key=%s""", ("types-002",))assert cur.fetchone() == (1, 1, 1, 1)Also test binary values with bytes against BINARY/VARBINARY/BLOB columns if your real schema uses them. Never infer type semantics from string rendering in logs.
Deterministic integration checks and plan visibility
SELECT work_order_id, estimated_cost, due_at, JSON_UNQUOTE(JSON_EXTRACT(metadata,'$.source')) AS source, JSON_LENGTH(JSON_EXTRACT(metadata,'$.flags')) AS flag_count, HEX(request_fingerprint) AS fingerprint_hexFROM servicehub_app_lab.work_ordersWHERE idempotency_key='types-001';SELECT idempotency_key, estimated_cost IS NULL AS decimal_is_null, due_at IS NULL AS temporal_is_null, metadata IS NULL AS json_is_null, request_fingerprint IS NULL AS binary_is_nullFROM servicehub_app_lab.work_ordersWHERE idempotency_key='types-002';EXPLAINSELECT work_order_id, status, priority, created_atFROM servicehub_app_lab.work_ordersWHERE status='open'ORDER BY created_at DESC, work_order_id DESCLIMIT 50;The test should assert stored values and business invariants, not only “no exception.” The plan evidence confirms what MySQL chose for that generated query; it does not prove every parameter value will have identical cardinality or cost.
Production judgment and bridge to Lesson 3
Use bound parameters for every untrusted data value and allowlists for dynamic SQL structure. Keep sensitive bind values out of broad production logs. Track driver errors, prepared-statement counts where relevant, query digests, and integration-test failures. The next lesson addresses a harder boundary: one logical business operation can span multiple SQL statements, failures, locks, and even lost connections.
Knowledge check
- Why should parameter markers not be quoted inside SQL text?
- Can a bound parameter represent a table name or DESC keyword?
- Why is Decimal preferable to float for many monetary fields?
- What should an injection-shaped integration test assert?
- Does a safe prepared statement guarantee a good query plan?
Reveal answers
- The connector owns conversion/binding. Quoting a marker turns it into SQL text or otherwise defeats the intended parameter protocol.
- No. Identifiers and SQL keywords are grammar; validate them against a developer-owned allowlist and interpolate only the mapped constant.
- DECIMAL is exact base-10 storage; Decimal preserves decimal intent while binary float can introduce representation error.
- That the supplied characters are treated as one literal value and cannot change row selection or statement structure.
- No. Security/correctness and optimizer access-path quality are separate concerns; inspect plans for important queries.