基于Python的应用开发（mysql-connector-python）

前提条件

• 已安装Python环境，且Python版本为3.8及以上版本。

• 已开通MySQL协议兼容功能。如何开通，请参见开通MySQL协议兼容功能。

• 已将客户端IP添加至白名单，具体操作请参见设置白名单。

操作步骤

1. 安装8.0.11版本的mysql-connector-python。您也可以通过pip install mysql-connector-python==8.0.11语句直接通过pip进行安装。

2. 创建连接并配置连接参数。

   connection = mysql.connector.connect(host='<MySQL兼容地址>', port=33060, user='<用户名>', passwd='<密码>', database='<数据库名>')
   cursor = connection.cursor(prepared=True)
   

   参数说明

   参数	说明
   host	Lindorm宽表引擎的MySQL兼容地址，需去掉末尾的冒号及端口号:33060。如何获取，请参见查看连接地址。 重要 如果应用部署在ECS，且ECS与Lindorm实例部署在同一专有网络，建议您通过专有网络访问Lindorm实例，否则请使用公网访问。通过公网连接Lindorm前需在控制台开通公网地址，开通方式请参见开通步骤。
   port	Lindorm宽表引擎MySQL协议的端口，固定为33060。
   user	如果您忘记用户密码，可以通过Lindorm宽表引擎的集群管理系统修改密码。具体操作，请参见修改用户密码。
   passwd
   database	需要连接的数据库名称。默认连接default数据库。

3. 通过宽表SQL语法使用Lindorm宽表引擎。以创建表为例。

   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)
   

完整示例

完整示例代码如下：

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

import mysql.connector

# 打开数据库连接。
# host是Lindorm宽表引擎MySQL协议的连接地址。
# port是Lindorm宽表引擎MySQL协议的端口，一般为33060。
# user是Lindorm宽表引擎的用户账号。
# passwd是Lindorm宽表引擎的用户账号对应的密码。
# database是Lindorm宽表引擎中的数据库名。


connection = mysql.connector.connect(host='ld-bp1hn6yq0yb34****-proxy-sql-lindorm-public.lindorm.rds.aliyuncs.com', port=33060,user='root', passwd='test',database='default')

# 创建cursor，注意prepared=True
cursor = connection.cursor(prepared=True)

# 创建表
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)

# 插入数据
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"})))

# 执行多条插入语句
data = [
    (3, 3, '3'),
    (4, 4, json.dumps({"key": "value4"})),
]
cursor.executemany(sql_upsert, data)

# 删除数据
sql_delete = "delete from test_python where c1 = ?"
print(sql_delete)
cursor.execute(sql_delete, (3,))

# 修改数据
sql_update = "upsert into test_python(c1, c2, c3) values(?, ?, ?)"
print(sql_update)
cursor.execute(sql_update, (1, 2, '2'))

# 查询特定数据
sql_select = "select * from test_python where c1 = ?"
print(sql_select)
cursor.execute(sql_select, (4,))
rows = cursor.fetchall()
print(rows)

# 查询表中所有数据
sql_select_all = "select * from test_python"
print(sql_select_all)
cursor.execute(sql_select_all)
rows = cursor.fetchall()
print(rows)

# 关闭 cursor
cursor.close()

# 关闭连接
connection.close()