Connecting Applications to MySQL
Applications talk to MySQL through connectors (drivers) that implement the MySQL client/server protocol (and, for some, the X Protocol on port 33060 used by MySQL Shell and the Document Store).
Concept
Applications talk to MySQL through connectors (drivers) that implement the MySQL client/server protocol (and, for some, the X Protocol on port 33060 used by MySQL Shell and the Document Store). Production applications also need connection pooling, timeouts, TLS, error handling with retries, and secure credential storage.
mysql-connector-python (Oracle), PyMySQL, mysqlclient; ORMs: SQLAlchemy, Django ORM
Connector/J (JDBC); pools: HikariCP; ORMs: Hibernate/JPA, jOOQ, MyBatis
mysql2; ORMs: Prisma, Sequelize, TypeORM, Knex PDO_MySQL, mysqli; Laravel Eloquent, Doctrine
MySqlConnector (open source), Connector/NET; Entity Framework Core (Pomelo provider)
go-sql-driver/mysql with database/sql; GORM, sqlx MySQL C API (libmysqlclient), Connector/C++ 8 (X DevAPI and JDBC-style API) mysql, sqlx
Why It Matters
Most production database problems seen by developers — "too many connections", timeouts, deadlocks surfacing as exceptions, garbled characters, leaked connections — are client-side configuration and coding issues.
Syntax
Typical connection parameters: host, port (3306), database, user, password, TLS mode, character set (utf8mb4), connect/read timeouts, time zone handling, pool size.
Example
35.1 Python with a connection pool and transactions 35.2 Java with HikariCP 35.3 Node.js with mysql2 pool 35.5 C# with MySqlConnector
35.6 C (MySQL C API) — for embedded Linux gateways and test benches
Compile: gcc can_writer.c $(mysql_config --cflags --libs) -o can_writer.
For embedded devices, it is usually better not to write every frame directly over the network: buffer locally (SQLite or files), then batch-upload (LOAD DATA LOCAL INFILE or multi-row inserts) to MySQL, so network drops do not lose data and the server is not flooded with tiny transactions.
35.7 Connection pooling guidelines Pool size per app instance is usually small (e.g., 5–20); total across all instances must stay below max_connections with headroom for admins and replicas.
Set pool maxLifetime below MySQL wait_timeout to avoid "server has gone away" errors.
Always return connections (try/finally, using, context managers).
For thousands of clients (serverless, many microservices), put a proxy in front: ProxySQL, MySQL Router, or managed proxies (RDS Proxy).
35.8 Time zones and character sets from applications Store times in UTC; set the session time zone explicitly (SET time_zone = '+00:00' or connector options).
Use utf8mb4 in the connection (charset=utf8mb4), matching the database.
import os
import mysql.connector
from mysql.connector import pooling, errorcode
pool = pooling.MySQLConnectionPool(
pool_name="eh_pool", pool_size=10,
host=os.environ["DB_HOST"], port=3306, database="eh_company",
user="eh_app", password=os.environ["DB_PASSWORD"],
charset="utf8mb4", ssl_disabled=False, autocommit=False,
)
def transfer_stock(src_product: int, dst_product: int, qty: int, retries: int = 3) -> None:
for attempt in range(retries):
conn = pool.get_connection()
try:
cur = conn.cursor()
conn.start_transaction()
cur.execute("UPDATE products SET stock = stock - %s WHERE product_id = %s AND stock >= %s",
(qty, src_product, qty))
if cur.rowcount != 1:
raise ValueError("insufficient stock or unknown product")
cur.execute("UPDATE products SET stock = stock + %s WHERE product_id = %s", (qty, dst_product))
if cur.rowcount != 1:
raise ValueError("unknown target product")
conn.commit()
return
except mysql.connector.Error as e:
conn.rollback()
if e.errno in (errorcod
…HikariConfig cfg = new HikariConfig();
cfg.setJdbcUrl("jdbc:mysql://db.internal:3306/eh_company"
+ "?sslMode=VERIFY_IDENTITY&connectionTimeZone=UTC&characterEncoding=UTF-8");
cfg.setUsername("eh_app");
cfg.setPassword(System.getenv("DB_PASSWORD"));
cfg.setMaximumPoolSize(20);
cfg.setConnectionTimeout(5_000);
cfg.addDataSourceProperty("cachePrepStmts", "true");
cfg.addDataSourceProperty("useServerPrepStmts", "true");
cfg.addDataSourceProperty("rewriteBatchedStatements", "true"); // faster batch inserts
HikariDataSource ds = new HikariDataSource(cfg);
try (Connection c = ds.getConnection()) {
c.setAutoCommit(false);
try (PreparedStatement ps = c.prepareStatement(
"INSERT INTO test_result (vehicle_id, sw_version_id, test_case, result, duration_s, executed_at) VALUES (?,?,?,?,?,?)")) {
for (TestRun r : runs) {
ps.setInt(1, r.vehicleId()); ps.setInt(2, r.swVersionId());
ps.setString(3, r.testCase()); ps.setString(4, r.result());
ps.setInt(5, r.durationS()); ps.setTimestamp(6, Timestamp.from(r.executedAt()));
ps.addBatch();
}
ps.executeBatch();
c.commit();
} catch (SQLExcept
…const mysql = require('mysql2/promise');
const pool = mysql.createPool({
host: process.env.DB_HOST, user: 'eh_app', password: process.env.DB_PASSWORD,
database: 'eh_company', connectionLimit: 10, charset: 'utf8mb4', timezone: 'Z',
});
async function ordersForCustomer(customerId) {
const [rows] = await pool.execute(
'SELECT order_id, order_date, total_amount FROM orders WHERE customer_id = ? ORDER BY order_date',
[customerId]);
return rows;
}$pdo = new PDO('mysql:host=db.internal;dbname=eh_company;charset=utf8mb4', 'eh_app', getenv('DB_PASSWORD'), [
PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
PDO::ATTR_EMULATE_PREPARES => false,
]);
$stmt = $pdo->prepare('SELECT name, city FROM customers WHERE country = :country');
$stmt->execute(['country' => 'India']);
foreach ($stmt as $row) { echo $row['name'], PHP_EOL; }await using var conn = new MySqlConnection(
"Server=db.internal;Database=eh_company;User ID=eh_app;Password=" + Environment.GetEnvironmentVariable("DB_PASSWORD") +
";SslMode=Required;CharSet=utf8mb4;Pooling=true;MaximumPoolSize=20");
await conn.OpenAsync();
await using var cmd = new MySqlCommand("SELECT COUNT(*) FROM orders WHERE status = @status", conn);
cmd.Parameters.AddWithValue("@status", "DELIVERED");
long delivered = (long)(await cmd.ExecuteScalarAsync())!; // 3#include <mysql.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
int main(void) {
MYSQL *db = mysql_init(NULL);
if (!mysql_real_connect(db, "db.internal", "bench_writer", getenv("DB_PASSWORD"),
"vehicle_test", 3306, NULL, 0)) {
fprintf(stderr, "connect: %s\n", mysql_error(db));
return 1;
}
mysql_set_character_set(db, "utf8mb4");
MYSQL_STMT *st = mysql_stmt_init(db);
const char *sql = "INSERT INTO can_frames (captured_at, can_id, dlc, payload) VALUES (NOW(6), ?, ?, ?)";
mysql_stmt_prepare(st, sql, strlen(sql));
unsigned int can_id = 0x1A0;
unsigned char dlc = 8;
unsigned char payload[8] = {0x0F, 0xA0, 0x00, 0x00, 0xC8, 0x00, 0x00, 0x00};
unsigned long payload_len = sizeof payload;
MYSQL_BIND b[3];
memset(b, 0, sizeof b);
b[0].buffer_type = MYSQL_TYPE_LONG; b[0].buffer = &can_id; b[0].is_unsigned = 1;
b[1].buffer_type = MYSQL_TYPE_TINY; b[1].buffer = &dlc; b[1].is_unsigned = 1;
b[2].buffer_type = MYSQL_TYPE_BLOB; b[2].buffer = payload; b[2].buffer_length = sizeof payload; b[2].length = &payload_len;
mysql_stmt_bind_param(st, b);
if (mysql_stmt
…Expected Result
The C# example returns 3 (orders 1001, 1002, 1006 are DELIVERED). The Node.js ordersForCustomer(2) returns orders 1002 and 1006.
Common Mistakes
Opening a new connection per query without pooling.
Not closing connections (pool exhaustion).
Autocommit assumptions: Python's mysql-connector defaults to autocommit=False — forgetting commit() loses writes.
String-built SQL (injection).
Ignoring deadlock errors instead of retrying.
Mismatched time zones between app and database.
Best Practices
Pool connections; set timeouts; retry transient errors (1213, 1205, lost connection on idempotent operations).
Parameterize everything; TLS on; secrets from a vault/environment.
Batch writes for telemetry and test data.
Interview Relevance
Medium: "How do you connect to MySQL from Java/Python?", "What is connection pooling?", "How do you handle deadlocks in application code?", "What does rewriteBatchedStatements do?".
Practice Questions
Write a Python script that bulk-inserts 10,000 test results using executemany and measure time vs single inserts.
Configure a pool whose connections are recycled before wait_timeout.
Implement deadlock retry logic in your language.