阿里云PolarDB是一种云原生关系型数据库，兼容MySQL、PostgreSQL等主流数据库协议，具备高可用、高并发、弹性扩容等特性，适用于电商、金融、社交等各类高负载业务场景。本文将围绕PolarDB MySQL版，使用Python SDK实现数据的增删改查（CRUD）核心操作，并针对高并发场景优化实现数据库连接池，附上完整代码示例与实战优化建议，帮助开发者快速上手PolarDB开发。
首先完成开发准备工作：1. 在阿里云控制台创建PolarDB MySQL版集群，获取数据库连接地址、端口、用户名、密码；2. 配置数据库白名单，将本地开发IP或函数计算的内网IP加入白名单，允许外部访问；3. 在Python环境中安装MySQL连接驱动，推荐使用PyMySQL，通过pip命令安装：pip install PyMySQL DBUtils，其中DBUtils用于实现数据库连接池，提升连接复用效率。
首先实现基础的数据库连接与CRUD操作，核心是通过PyMySQL建立数据库连接，执行SQL语句，处理查询结果与异常。以下是基础CRUD的代码示例，包含用户表的创建、用户信息的新增、查询、更新、删除等操作：
import pymysql
from pymysql.cursors import DictCursor

配置PolarDB MySQL连接信息

POLARDB_CONFIG = {
"host": "pc-xxxxxx.mysql.polaradb.aliyuncs.com", # 替换为你的PolarDB连接地址
"port": 3306, # PolarDB MySQL默认端口
"user": "your_username", # 替换为数据库用户名
"password": "your_password", # 替换为数据库密码
"database": "test_db", # 替换为你的数据库名称
"charset": "utf8mb4",
"autocommit": True, # 自动提交事务
"cursorclass": DictCursor # 返回字典格式的查询结果
}

def get_db_connection():
"""获取数据库连接"""
try:
conn = pymysql.connect(**POLARDB_CONFIG)
print("成功连接到PolarDB MySQL数据库")
return conn
except pymysql.MySQLError as e:
print(f"数据库连接失败：{e.args[0]} - {e.args[1]}")
return None

def create_user_table():
"""创建用户表"""
conn = get_db_connection()
if not conn:
return False
try:
with conn.cursor() as cursor:

        # 定义创建表的SQL语句
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS users (
            id INT PRIMARY KEY AUTO_INCREMENT COMMENT '用户ID',
            name VARCHAR(50) NOT NULL COMMENT '用户名',
            age INT COMMENT '用户年龄',
            email VARCHAR(100) UNIQUE COMMENT '用户邮箱',
            create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
            update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间'
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户信息表';
        """
        cursor.execute(create_table_sql)
        print("用户表创建成功（或已存在）")
        return True
except pymysql.MySQLError as e:
    print(f"创建用户表失败：{e.args[0]} - {e.args[1]}")
    return False
finally:
    if conn:
        conn.close()  # 关闭数据库连接

def add_user(name, age, email):
"""新增用户信息"""
conn = get_db_connection()
if not conn:
return False, "数据库连接失败"
try:
with conn.cursor() as cursor:

        # 定义插入SQL语句
        insert_sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s);"
        affected_rows = cursor.execute(insert_sql, (name, age, email))
        if affected_rows > 0:
            user_id = cursor.lastrowid
            return True, f"新增用户成功，用户ID：{user_id}"
        else:
            return False, "新增用户失败，无数据插入"
except pymysql.MySQLError as e:
    err_msg = f"新增用户失败：{e.args[0]} - {e.args[1]}"
    print(err_msg)
    return False, err_msg
finally:
    if conn:
        conn.close()

def get_user_by_id(user_id):
"""根据用户ID查询用户信息"""
conn = get_db_connection()
if not conn:
return None, "数据库连接失败"
try:
with conn.cursor() as cursor:
select_sql = "SELECT * FROM users WHERE id = %s;"
cursor.execute(select_sql, (user_id,))
user = cursor.fetchone()
if user:
return user, "查询成功"
else:
return None, f"用户ID {user_id} 不存在"
except pymysql.MySQLError as e:
err_msg = f"查询用户失败：{e.args[0]} - {e.args[1]}"
print(err_msg)
return None, err_msg
finally:
if conn:
conn.close()

def update_user(user_id, **kwargs):
"""更新用户信息"""
conn = get_db_connection()
if not conn:
return False, "数据库连接失败"
try:

    # 构建更新字段与参数
    update_fields = []
    params = []
    for key, value in kwargs.items():
        if key in ["name", "age", "email"]:  # 仅允许更新指定字段
            update_fields.append(f"{key} = %s")
            params.append(value)
    params.append(user_id)
    if not update_fields:
        return False, "未指定需要更新的字段"
    update_sql = f"UPDATE users SET {', '.join(update_fields)} WHERE id = %s;"
    with conn.cursor() as cursor:
        affected_rows = cursor.execute(update_sql, params)
        if affected_rows > 0:
            return True, "用户信息更新成功"
        else:
            return False, f"用户ID {user_id} 不存在或无字段更新"
except pymysql.MySQLError as e:
    err_msg = f"更新用户失败：{e.args[0]} - {e.args[1]}"
    print(err_msg)
    return False, err_msg
finally:
    if conn:
        conn.close()

def delete_user(user_id):
"""删除用户信息"""
conn = get_db_connection()
if not conn:
return False, "数据库连接失败"
try:
with conn.cursor() as cursor:
delete_sql = "DELETE FROM users WHERE id = %s;"
affected_rows = cursor.execute(delete_sql, (user_id,))
if affected_rows > 0:
return True, "用户信息删除成功"
else:
return False, f"用户ID {user_id} 不存在"
except pymysql.MySQLError as e:
err_msg = f"删除用户失败：{e.args[0]} - {e.args[1]}"
print(err_msg)
return False, err_msg
finally:
if conn:
conn.close()

测试基础CRUD操作

if name == "main":

# 创建用户表
create_user_table()
# 新增用户
add_result, add_msg = add_user("张三", 25, "zhangsan@example.com")
print(add_msg)
# 查询用户
user, query_msg = get_user_by_id(1)
print(query_msg, user)
# 更新用户
update_result, update_msg = update_user(1, age=26, email="zhangsan_new@example.com")
print(update_msg)
# 删除用户
delete_result, delete_msg = delete_user(1)
print(delete_msg)

在高并发业务场景中，频繁创建和关闭数据库连接会导致性能瓶颈，此时需要使用数据库连接池优化连接管理。DBUtils是Python中常用的数据库连接池工具，分为PersistentDB（线程专用连接池）和PooledDB（线程共享连接池），本文使用PooledDB实现连接池，实现连接复用，减少连接创建开销。
以下是基于连接池的优化代码，替换基础的连接获取方式，实现连接池的初始化、连接获取与释放：
import pymysql
from pymysql.cursors import DictCursor
from DBUtils.PooledDB import PooledDB

配置PolarDB MySQL连接信息（同基础操作）

POLARDB_CONFIG = {
"host": "pc-xxxxxx.mysql.polaradb.aliyuncs.com",
"port": 3306,
"user": "your_username",
"password": "your_password",
"database": "test_db",
"charset": "utf8mb4",
"autocommit": True,
"cursorclass": DictCursor
}

初始化数据库连接池

pool = PooledDB(
creator=pymysql, # 使用的数据库驱动
maxconnections=10, # 连接池最大连接数
mincached=2, # 初始化时连接池中至少创建的空闲连接数
maxcached=5, # 连接池中最多闲置的连接数
maxshared=3, # 连接池中最多共享的连接数
blocking=True, # 连接数达到最大时，是否阻塞等待
maxusage=None, # 单个连接最多被使用的次数，None表示无限制
setsession=[], # 初始化连接时执行的SQL语句
ping=0, # 检查连接可用性的方式，0表示不检查
**POLARDB_CONFIG
)

def get_db_connection_from_pool():
"""从连接池获取数据库连接"""
try:
conn = pool.connection() # 从连接池获取连接
print("从连接池成功获取数据库连接")
return conn
except Exception as e:
print(f"从连接池获取连接失败：{str(e)}")
return None

基于连接池的新增用户函数（其他CRUD函数可同理修改）

def add_user_with_pool(name, age, email):
"""使用连接池新增用户信息"""
conn = get_db_connection_from_pool()
if not conn:
return False, "从连接池获取连接失败"
try:
with conn.cursor() as cursor:
insert_sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s);"
affected_rows = cursor.execute(insert_sql, (name, age, email))
if affected_rows > 0:
user_id = cursor.lastrowid
return True, f"新增用户成功，用户ID：{user_id}"
else:
return False, "新增用户失败，无数据插入"
except pymysql.MySQLError as e:
err_msg = f"新增用户失败：{e.args[0]} - {e.args[1]}"
print(err_msg)
return False, err_msg
finally:
if conn:
conn.close() # 将连接归还到连接池，而非真正关闭

测试连接池操作

if name == "main":
create_user_table() # 先创建用户表

# 使用连接池新增多个用户
for i in range(5):
    name = f"测试用户{i+1}"
    email = f"test{i+1}@example.com"
    result, msg = add_user_with_pool(name, 20+i, email)
    print(msg)
# 使用连接池查询用户
user, msg = get_user_by_id(2)
print(msg, user)

在实际开发中，还需要注意以下优化与安全事项：1. 连接池参数需要根据业务并发量进行调整，maxconnections不宜过大或过小，避免资源浪费或连接不足；2. 使用参数化SQL语句，避免SQL注入攻击，本文中所有SQL语句均使用参数化方式，提升安全性；3. 配置PolarDB的读写分离，针对查询请求路由到只读节点，提升查询性能；4. 引入数据库事务，针对多步操作（如转账、订单创建），保证数据一致性；5. 收集数据库慢查询日志，使用阿里云DMS工具分析慢查询，优化SQL语句与表结构。
阿里云PolarDB MySQL版兼容MySQL协议，开发者可以快速将现有MySQL应用迁移到PolarDB，无需修改大量代码。通过连接池优化，能够有效提升高并发场景下的数据库性能，减少连接开销。后续可结合PolarDB的弹性扩容、自动备份、高可用等特性，构建稳定、高效的数据库服务，支撑各类业务的发展。

阿里云PolarDB #MySQL #Python SDK #数据库连接池 #CRUD实战

阿里云PolarDB MySQL版Python SDK实战——实现数据增删改查与连接池优化

配置PolarDB MySQL连接信息

测试基础CRUD操作

配置PolarDB MySQL连接信息（同基础操作）

初始化数据库连接池

基于连接池的新增用户函数（其他CRUD函数可同理修改）

测试连接池操作

阿里云PolarDB #MySQL #Python SDK #数据库连接池 #CRUD实战

热门文章

最新文章

相关电子书

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

阿里云PolarDB MySQL版Python SDK实战——实现数据增删改查与连接池优化

配置PolarDB MySQL连接信息

测试基础CRUD操作

配置PolarDB MySQL连接信息（同基础操作）

初始化数据库连接池

基于连接池的新增用户函数（其他CRUD函数可同理修改）

测试连接池操作

阿里云PolarDB #MySQL #Python SDK #数据库连接池 #CRUD实战

热门文章

最新文章

相关电子书