文档反馈官招募中，报名立赚积分兑换代金券！> HOT
文档中心>TDSQL Boundless>开发指南>开发指南（MySQL 兼容模式）>示例程序>Python

Python

最近更新时间：2026-02-05 11:01:13

我的收藏

本页目录：

概述

TDSQL Boundless 高度兼容 MySQL 协议，所有支持 MySQL 的 Python 数据库驱动均能适配 TDSQL Boundless。本文详细梳理主流驱动、特性、使用示例及 TDSQL Boundless 适配注意事项。
驱动名称
类型
核心优势
适用场景
mysql-connector-python
官方驱动
由 Oracle/MySQL 官方维护，兼容性最佳，无额外依赖
追求稳定性、无需复杂配置的场景

PyMySQL
纯 Python 实现
轻量灵活、源码易读、安装简单、社区活跃
快速开发、小型项目、需要自定义扩展的场景
mysqlclient
C 语言扩展
性能最优（比纯 Python 驱动快20%-50%）
高并发 OLTP 场景、大数据量读写场景
aiomysql
异步纯 Python
支持 asyncio 异步 IO，高并发异步场景首选
FastAPI、aiohttp 等异步 Web 框架项目
SQLAlchemy
ORM 框架（依赖底层驱动）
屏蔽底层驱动差异、支持 SQL 抽象与事务管理
中大型项目、需要跨数据库兼容的场景

mysql-connector-python（官方首选，稳定兼容）

特点

MySQL 官方出品，与 TDSQL Boundless（MySQL 兼容）的协议兼容性无死角。
纯 Python 实现，无需依赖系统 MySQL 库，跨平台部署便捷。
支持 SSL 连接、批量操作、事务管理等核心功能，适配 TDSQL Boundless 金融级安全需求。

TDSQL Boundless 连接示例

import mysql.connector

# 配置数据库连接信息（替换为你的实际配置）
db_config = {
    "host": "10.10.10.10",
    "user": "test",
    "port": 7981,
    "password": "test123",
    "database": "test"
}
# 建立连接 + 创建单个游标（复用该游标执行所有SQL）
conn = mysql.connector.connect(**db_config)
cursor = conn.cursor(buffered=True)

# 1. 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS test_table (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    age TINYINT DEFAULT 0
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""")

# 2. 插入一行数据并提交事务
cursor.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ("Tom", 22))
conn.commit()

# 3. 查询数据并打印
cursor.execute("SELECT * FROM test_table WHERE name = %s", ("Tom",))
data = cursor.fetchone()
print("Query result:", data)

# 关闭游标和连接
cursor.close()
conn.close()

PyMySQL（轻量灵活，快速开发）

特点

纯 Python 实现，安装无依赖，新手友好。
语法与 mysql-connector-python 高度兼容，迁移成本低。
支持连接池、批量操作，社区维护活跃，问题排查便捷。

TDSQL Boundless 连接示例

import pymysql

# 配置数据库连接信息（替换为你的实际配置）
db_config = {
    "host": "10.10.10.10",
    "user": "test",
    "port": 7981,
    "password": "test123",
    "database": "test"
}

# 建立连接 + 创建游标
conn = pymysql.connect(**db_config)
cursor = conn.cursor()

# 1. 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS test_table (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    age TINYINT DEFAULT 0
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""")

# 2. 插入一行数据并提交事务
cursor.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ("Tom", 22))
conn.commit()

# 3. 查询数据并打印
cursor.execute("SELECT * FROM test_table WHERE name = %s", ("Tom",))
data = cursor.fetchone()
print("Query result:", data)

# 关闭游标和连接
cursor.close()
conn.close()

mysqlclient（高性能首选，高并发场景）

特点

C 语言封装 MySQL 客户端库，性能远超纯 Python 驱动。
兼容 DB API 2.0 规范，是 Django、SQLAlchemy 的默认推荐驱动。
支持大批量数据读写、复杂 SQL 执行，高度适配 TDSQL Boundless。

TDSQL Boundless 连接示例

import MySQLdb

# 配置数据库连接信息（替换为你的实际配置）
db_config = {
    "host": "10.10.10.10",
    "user": "test",
    "port": 7981,
    "password": "test123",
    "database": "test"
}


# 建立连接 + 创建游标
conn = MySQLdb.connect(**db_config)
cursor = conn.cursor()

# 1. 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS test_table (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    age TINYINT DEFAULT 0
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""")

# 2. 插入一行数据并提交事务
cursor.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ("Tom", 22))
conn.commit()

# 3. 查询数据并打印
cursor.execute("SELECT * FROM test_table WHERE name = %s", ("Tom",))
data = cursor.fetchone()
print("Query result:", data)

# 关闭游标和连接
cursor.close()
conn.close()

aiomysql（异步首选，异步高并发场景）

特点

基于 PyMySQL 封装，支持 asyncio 异步 IO，无阻塞连接。
适配 FastAPI、aiohttp 等异步框架，单进程支持上万并发连接。
支持异步连接池、异步事务，适配 TDSQL Boundless 分布式高可用场景。

TDSQL Boundless 连接示例

import asyncio
import aiomysql

# 配置数据库连接信息（替换为你的实际配置）
db_config = {
    "host": "10.10.10.10",
    "user": "test",
    "port": 7981,
    "password": "test123",
    "db": "test"
}


async def main():
    # 1. 建立异步连接
    conn = await aiomysql.connect(**db_config)
    # 2. 创建异步游标
    cursor = await conn.cursor()

    # 3. 创建表
    await cursor.execute("""
    CREATE TABLE IF NOT EXISTS test_table (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(50) NOT NULL,
        age TINYINT DEFAULT 0
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)

    # 4. 插入一行数据并提交事务
    await cursor.execute("INSERT INTO test_table (name, age) VALUES (%s, %s)", ("Tom", 22))
    await conn.commit()

    # 5. 查询数据并打印
    await cursor.execute("SELECT * FROM test_table WHERE name = %s", ("Tom",))
    data = await cursor.fetchone()
    print("Query result:", data)

    # 6. 关闭游标和连接
    await cursor.close()
    conn.close()

# Python 3.6 兼容的异步启动方式（替换 asyncio.run()）
if __name__ == "__main__":
    # 1. 获取默认事件循环
    loop = asyncio.get_event_loop()
    # 2. 运行异步主函数
    loop.run_until_complete(main())
    # 3. 关闭事件循环
    loop.close()

SQLAlchemy（ORM 框架，中大型项目首选）

特点

高层 ORM 框架，屏蔽底层驱动差异（可兼容 mysqlclient/PyMySQL 等）。
支持 SQL 抽象、事务管理、模型映射，减少重复 SQL 编码。
适配 TDSQL Boundless 分表、读写分离场景，支持动态切换数据源。

TDSQL Boundless 连接示例

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 1. 基础配置：创建基类 + 数据库连接引擎
Base = declarative_base()
# 连接字符串格式：mysql+底层驱动://账号:密码@地址:端口/数据库名?字符集
engine = create_engine("mysql+mysqldb://test:test123@10.10.10.10:7981/test?charset=utf8mb4")

# 2. 定义 ORM 模型（对应 test_table 表）
class TestTable(Base):
    __tablename__ = "test_table"  # 表名
    id = Column(Integer, primary_key=True, autoincrement=True)  # 自增主键
    name = Column(String(50), nullable=False)  # 用户名
    age = Column(Integer, default=0)  # 年龄

# 3. 创建会话（用于数据库操作）
Session = sessionmaker(bind=engine)
session = Session()

# 4. 创建表（若不存在）
Base.metadata.create_all(engine)

# 5. 插入一行数据
new_data = TestTable(name="Tom", age=22)
session.add(new_data)
session.commit()  # 提交事务

# 6. 查询数据并打印
query_data = session.query(TestTable).filter(TestTable.name == "Tom").first()
print("Query result:", (query_data.id, query_data.name, query_data.age))

# 7. 关闭会话
session.close()
中国站
目录