阿里云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实战