Connect to LindormTable Using mysql-connector-python - Lindorm

mysql-connector-python is the official pure-Python MySQL connector that requires no C library dependencies. This guide shows you how to connect a Python application to LindormTable and run CRUD operations. By the end, you will be able to:

Install and configure the connector
Connect to LindormTable in direct connection mode or connection pool mode
Create, insert, update, delete, and query data using LindormTable's wide table SQL syntax

Prerequisites

Before you begin, ensure that you have:

Python 3.8 or later installed
MySQL compatibility enabled for your Lindorm instance. For details, see Enable the MySQL compatibility feature
Your client IP address added to the instance whitelist. For details, see Configure whitelists

Install the connector

Run the following command to install mysql-connector-python 8.0.15:

pip install mysql-connector-python==8.0.15

Connect to LindormTable

The following table describes the connection parameters used in all examples. Replace the placeholder values with your actual values before running any code.

Parameter	Description	Example
`host`	The Lindorm Wide Table SQL Address, with the colon and port number (`:33060`) removed. To find this value, see View endpoints.	`ld-bp1hn6yq0yb34****-proxy-sql-lindorm-public.lindorm.rds.aliyuncs.com`
`port`	Fixed at `33060`.	`33060`
`user`	Your LindormTable username. To reset a forgotten password, see Change your password.	`root`
`passwd`	Your LindormTable password.	—
`database`	The database to connect to. Defaults to `default` if not specified.	`default`

Important

If your application runs on an Elastic Compute Service (ECS) instance in the same Virtual Private Cloud (VPC) as your Lindorm instance, connect through the VPC endpoint. Otherwise, connect over the public endpoint. To enable the public endpoint, see Activate LindormSearch.

Important

Always create the cursor with prepared=True. Without it, strings containing special characters such as double quotation marks (") may be incorrectly escaped when written to LindormTable.

Choose a connection mode

mysql-connector-python supports two connection modes:

Mode	When to use
Direct connection	One-time operations or infrequent access. Opens and closes a new connection for each operation.
Connection pool	Frequent access. Reuses connections to reduce overhead and improve performance.

Code examples

Direct connection mode

The following example establishes a direct connection and runs create, insert, delete, update, and query operations on a test table.

#!/usr/bin/python
# -*- coding: UTF-8 -*-
import json

import mysql.connector

# Establish a connection. Replace the placeholder values with your own.
connection = mysql.connector.connect(
    host='<Lindorm Wide Table SQL Address>',
    port=33060,
    user='<username>',
    passwd='<password>',
    database='default'
)

# Create a cursor. Use prepared=True to handle special characters correctly.
cursor = connection.cursor(prepared=True)

# Create a table.
sql_create_table = (
    "create table if not exists test_python"
    "(c1 integer, c2 integer, c3 varchar, primary key(c1))"
)
print(sql_create_table)
cursor.execute(sql_create_table)

# Insert data — single rows.
sql_upsert = "upsert into test_python(c1, c2, c3) values(?, ?, ?)"
print(sql_upsert)
cursor.execute(sql_upsert, (1, 1, '1'))
cursor.execute(sql_upsert, (2, 2, json.dumps({"key": "value2"})))

# Insert data — two rows in a batch.
sql_upsert_batch = "upsert into test_python(c1, c2, c3) values(?, ?, ?), (?, ?, ?)"
cursor.execute(sql_upsert_batch, (3, 3, '3', 4, 4, json.dumps({"key": "value4"})))

# Delete data.
sql_delete = "delete from test_python where c1 = ?"
print(sql_delete)
cursor.execute(sql_delete, (3,))

# Update data.
sql_update = "upsert into test_python(c1, c2, c3) values(?, ?, ?)"
print(sql_update)
cursor.execute(sql_update, (1, 2, '2'))

# Query a specific row.
sql_select = "select * from test_python where c1 = ?"
print(sql_select)
cursor.execute(sql_select, (4,))
rows = cursor.fetchall()
print(rows)

# Query all rows.
sql_select_all = "select * from test_python"
print(sql_select_all)
cursor.execute(sql_select_all)
rows = cursor.fetchall()
print(rows)

# Close the cursor and connection.
cursor.close()
connection.close()

Connection pool mode

The following example creates a connection pool, retrieves a connection from it, and runs the same set of operations. When you call connection.close(), the connection returns to the pool instead of closing permanently.

#!/usr/bin/python
# -*- coding: UTF-8 -*-
import json

import mysql.connector
from mysql.connector import pooling

# Create a connection pool.
# Adjust pool_size based on your expected workload concurrency.
connection_pool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=20,
    host='<Lindorm Wide Table SQL Address>',
    port=33060,
    user='<username>',
    password='<password>',
    database='default',
)

# Get a connection from the pool.
connection = connection_pool.get_connection()

# Create a cursor. Use prepared=True to handle special characters correctly.
cursor = connection.cursor(prepared=True)

# Drop the table if it exists, then recreate it.
sql_drop_table = "drop table if exists test_python"
print(sql_drop_table)
cursor.execute(sql_drop_table)

sql_create_table = (
    "create table test_python"
    "(c1 integer, c2 integer, c3 varchar, primary key(c1))"
)
print(sql_create_table)
cursor.execute(sql_create_table)

# Insert data — single rows.
sql_upsert = "insert into test_python(c1, c2, c3) values(?, ?, ?)"
print(sql_upsert)
cursor.execute(sql_upsert, (1, 1, '1'))
cursor.execute(sql_upsert, (2, 2, '2'))

# Insert data — three rows in a batch.
sql_upsert_batch = (
    "insert into test_python(c1, c2, c3) values(?, ?, ?), (?, ?, ?), (?, ?, ?)"
)
cursor.execute(sql_upsert_batch, (3, 3, '3', 4, 4, '4', 5, 5, '5'))

# Delete data.
sql_delete = "delete from test_python where c1 = ?"
print(sql_delete)
cursor.execute(sql_delete, (3,))

# Update data.
sql_update = "upsert into test_python(c1, c2, c3) values(?, ?, ?)"
print(sql_update)
cursor.execute(sql_update, (1, 2, '2'))

# Query a specific row.
sql_select = "select * from test_python where c1 = ?"
print(sql_select)
cursor.execute(sql_select, (4,))
rows = cursor.fetchall()
print(rows)

# Query all rows.
sql_select_all = "select * from test_python"
print(sql_select_all)
cursor.execute(sql_select_all)
rows = cursor.fetchall()
print(rows)

# Close the cursor.
# Calling connection.close() returns the connection to the pool.
cursor.close()
connection.close()

What's next

View endpoints — Find the correct endpoint for VPC or public access
Change your password — Reset your LindormTable credentials