MySQL Protocol Examples

Comprehensive code examples for using HeliosDB with MySQL protocol.

Connection Examples

Python with mysql-connector-python

import mysql.connector
from mysql.connector import pooling

# Basic connection
def basic_connection():
    conn = mysql.connector.connect(
        host="localhost",
        port=3306,
        user="admin",
        password="password",
        database="heliosdb"
    )
    return conn

# Connection pool
def create_pool():
    pool = pooling.MySQLConnectionPool(
        pool_name="mypool",
        pool_size=10,
        host="localhost",
        port=3306,
        user="admin",
        password="password",
        database="heliosdb"
    )
    return pool

# SSL connection
def ssl_connection():
    conn = mysql.connector.connect(
        host="localhost",
        port=3306,
        user="admin",
        password="password",
        database="heliosdb",
        ssl_ca="/path/to/ca.pem",
        ssl_cert="/path/to/client-cert.pem",
        ssl_key="/path/to/client-key.pem"
    )
    return conn

Node.js with mysql2

const mysql = require('mysql2/promise');

// Basic connection
async function basicConnection() {
  const connection = await mysql.createConnection({
    host: 'localhost',
    port: 3306,
    user: 'admin',
    password: 'password',
    database: 'heliosdb'
  });
  return connection;
}

// Connection pool
function createPool() {
  return mysql.createPool({
    host: 'localhost',
    port: 3306,
    user: 'admin',
    password: 'password',
    database: 'heliosdb',
    waitForConnections: true,
    connectionLimit: 10,
    queueLimit: 0
  });
}

// Execute query
async function executeQuery() {
  const pool = createPool();
  const [rows, fields] = await pool.execute(
    'SELECT * FROM users WHERE status = ?',
    ['active']
  );
  return rows;
}

Go with go-sql-driver/mysql

package main

import (
    "database/sql"
    "fmt"
    _ "github.com/go-sql-driver/mysql"
)

func main() {
    // Connection string
    dsn := "admin:password@tcp(localhost:3306)/heliosdb"

    // Open connection
    db, err := sql.Open("mysql", dsn)
    if err != nil {
        panic(err)
    }
    defer db.Close()

    // Configure pool
    db.SetMaxOpenConns(25)
    db.SetMaxIdleConns(5)

    // Query
    rows, err := db.Query("SELECT id, name FROM users")
    if err != nil {
        panic(err)
    }
    defer rows.Close()

    for rows.Next() {
        var id int
        var name string
        rows.Scan(&id, &name)
        fmt.Printf("ID: %d, Name: %s\n", id, name)
    }
}

CRUD Operations

Create (INSERT)

import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    port=3306,
    user="admin",
    password="password",
    database="heliosdb"
)
cursor = conn.cursor()

# Single insert
cursor.execute("""
    INSERT INTO users (name, email, created_at)
    VALUES (%s, %s, NOW())
""", ("Alice", "alice@example.com"))

# Bulk insert
users = [
    ("Bob", "bob@example.com"),
    ("Charlie", "charlie@example.com"),
    ("Diana", "diana@example.com")
]
cursor.executemany("""
    INSERT INTO users (name, email)
    VALUES (%s, %s)
""", users)

# Upsert (ON DUPLICATE KEY UPDATE)
cursor.execute("""
    INSERT INTO users (id, name, email, updated_at)
    VALUES (%s, %s, %s, NOW())
    ON DUPLICATE KEY UPDATE
        name = VALUES(name),
        email = VALUES(email),
        updated_at = NOW()
""", (1, "Alice Updated", "alice.new@example.com"))

conn.commit()
cursor.close()
conn.close()

Read (SELECT)

import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    port=3306,
    user="admin",
    password="password",
    database="heliosdb"
)
cursor = conn.cursor(dictionary=True)  # Return rows as dictionaries

# Simple query
cursor.execute("SELECT * FROM users WHERE status = 'active'")
users = cursor.fetchall()

# With parameters
cursor.execute("""
    SELECT u.*, COUNT(o.id) as order_count
    FROM users u
    LEFT JOIN orders o ON u.id = o.user_id
    WHERE u.created_at > %s
    GROUP BY u.id
    HAVING order_count > %s
    ORDER BY order_count DESC
    LIMIT %s
""", ("2024-01-01", 5, 10))

top_users = cursor.fetchall()
for user in top_users:
    print(f"{user['name']}: {user['order_count']} orders")

cursor.close()
conn.close()

Update

import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    port=3306,
    user="admin",
    password="password",
    database="heliosdb"
)
cursor = conn.cursor()

# Single update
cursor.execute("""
    UPDATE users
    SET status = %s, updated_at = NOW()
    WHERE id = %s
""", ("inactive", 1))

# Bulk update with conditions
cursor.execute("""
    UPDATE orders
    SET status = 'cancelled'
    WHERE created_at < DATE_SUB(NOW(), INTERVAL 30 DAY)
      AND status = 'pending'
""")

print(f"Rows affected: {cursor.rowcount}")
conn.commit()
cursor.close()
conn.close()

Delete

import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    port=3306,
    user="admin",
    password="password",
    database="heliosdb"
)
cursor = conn.cursor()

# Soft delete
cursor.execute("""
    UPDATE users
    SET deleted_at = NOW(), status = 'deleted'
    WHERE id = %s
""", (1,))

# Hard delete
cursor.execute("""
    DELETE FROM sessions
    WHERE expires_at < NOW()
""")

print(f"Rows deleted: {cursor.rowcount}")
conn.commit()
cursor.close()
conn.close()

Transactions

import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    port=3306,
    user="admin",
    password="password",
    database="heliosdb",
    autocommit=False
)
cursor = conn.cursor()

try:
    # Start transaction
    conn.start_transaction(isolation_level='REPEATABLE READ')

    # Debit from account A
    cursor.execute("""
        UPDATE accounts SET balance = balance - %s
        WHERE id = %s AND balance >= %s
    """, (100, 1, 100))

    if cursor.rowcount == 0:
        raise Exception("Insufficient funds")

    # Credit to account B
    cursor.execute("""
        UPDATE accounts SET balance = balance + %s
        WHERE id = %s
    """, (100, 2))

    # Record transfer
    cursor.execute("""
        INSERT INTO transfers (from_account, to_account, amount, created_at)
        VALUES (%s, %s, %s, NOW())
    """, (1, 2, 100))

    # Commit transaction
    conn.commit()
    print("Transfer successful")

except Exception as e:
    # Rollback on error
    conn.rollback()
    print(f"Transfer failed: {e}")
finally:
    cursor.close()
    conn.close()

Prepared Statements

import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    port=3306,
    user="admin",
    password="password",
    database="heliosdb"
)
cursor = conn.cursor(prepared=True)

# Prepare statement
stmt = """
    SELECT id, name, email
    FROM users
    WHERE status = %s AND created_at > %s
"""

# Execute multiple times with different parameters
for status in ['active', 'pending', 'inactive']:
    cursor.execute(stmt, (status, '2024-01-01'))
    users = cursor.fetchall()
    print(f"{status}: {len(users)} users")

cursor.close()
conn.close()

JSON Operations

import mysql.connector
import json

conn = mysql.connector.connect(
    host="localhost",
    port=3306,
    user="admin",
    password="password",
    database="heliosdb"
)
cursor = conn.cursor(dictionary=True)

# Create table with JSON column
cursor.execute("""
    CREATE TABLE IF NOT EXISTS products (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(255),
        attributes JSON
    )
""")

# Insert JSON data
attributes = {
    "color": "red",
    "size": "large",
    "tags": ["sale", "new", "featured"]
}
cursor.execute("""
    INSERT INTO products (name, attributes)
    VALUES (%s, %s)
""", ("T-Shirt", json.dumps(attributes)))

# Query JSON fields
cursor.execute("""
    SELECT
        name,
        JSON_EXTRACT(attributes, '$.color') as color,
        JSON_EXTRACT(attributes, '$.tags[0]') as first_tag
    FROM products
    WHERE JSON_CONTAINS(attributes, '"sale"', '$.tags')
""")

for product in cursor.fetchall():
    print(f"{product['name']}: {product['color']}")

# Update JSON
cursor.execute("""
    UPDATE products
    SET attributes = JSON_SET(attributes, '$.price', 29.99)
    WHERE id = %s
""", (1,))

conn.commit()
cursor.close()
conn.close()

Bulk Loading

import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    port=3306,
    user="admin",
    password="password",
    database="heliosdb",
    allow_local_infile=True
)
cursor = conn.cursor()

# Load data from CSV file
cursor.execute("""
    LOAD DATA LOCAL INFILE '/path/to/data.csv'
    INTO TABLE users
    FIELDS TERMINATED BY ','
    ENCLOSED BY '"'
    LINES TERMINATED BY '\n'
    IGNORE 1 ROWS
    (name, email, created_at)
""")

print(f"Loaded {cursor.rowcount} rows")
conn.commit()
cursor.close()
conn.close()

Error Handling

import mysql.connector
from mysql.connector import Error, IntegrityError, DatabaseError

def safe_query():
    try:
        conn = mysql.connector.connect(
            host="localhost",
            port=3306,
            user="admin",
            password="password",
            database="heliosdb"
        )
        cursor = conn.cursor()

        cursor.execute("INSERT INTO users (email) VALUES (%s)", ("duplicate@example.com",))
        conn.commit()

    except IntegrityError as e:
        print(f"Constraint violation: {e}")
        conn.rollback()
    except DatabaseError as e:
        print(f"Database error: {e}")
        conn.rollback()
    except Error as e:
        print(f"MySQL error: {e}")
    finally:
        if 'cursor' in locals():
            cursor.close()
        if 'conn' in locals():
            conn.close()

Last Updated: January 2026