Python 连接 MySQL 数据库

Python 连接 MySQL 数据库

在 python 中连接 MySQL 数据库，最常用、最稳定的方法是使用 PyMySQL 库。

一、准备工作

1.安装 PyMySQL 库

pipinstallpymysql

2.前提条件

• 本地 / 远程已安装 MySQL 服务
• 已创建数据库
• 拥有 MySQL 账号密码
• 确保 MySQL 端口（默认3306）未被防火墙拦截

二、连接 MySQL

# 1. 导入pymysql库importpymysql# 2. 创建数据库连接（核心代码）conn=pymysql.connect(host="localhost",# 数据库地址（本地用localhost/127.0.0.1）user="root",# MySQL用户名password="123456",# MySQL密码database="test_db",# 要连接的数据库名port=3306,# MySQL默认端口charset="utf8mb4"# 字符集（支持emoji，推荐）)# 3. 创建游标对象（用于执行SQL语句）cursor=conn.cursor()# 4. 执行SQL查询语句sql="SELECT * FROM user"# 替换为你的表名cursor.execute(sql)# 5. 获取查询结果# fetchone()：获取1条数据 | fetchall()：获取所有数据results=cursor.fetchall()forrowinresults:print(row)# 打印每一行数据# 6. 关闭游标和连接（必须！释放资源）cursor.close()conn.close()

三、核心操作

查询：无需提交事务

增 / 删 / 改：必须执行conn.commit()提交事务，否则数据不会生效！

1.插入数据

importpymysql conn=pymysql.connect(host="localhost",user="root",password="123456",database="test_db",charset="utf8mb4")cursor=conn.cursor()# 插入SQL（占位符 %s 防止SQL注入，推荐写法）sql="INSERT INTO user(name, age) VALUES (%s, %s)"data=("张三",20)# 数据元组try:cursor.execute(sql,data)conn.commit()# 提交事务！print("插入成功")exceptExceptionase:conn.rollback()# 出错回滚print("插入失败：",e)cursor.close()conn.close()

2.更新数据

sql="UPDATE user SET age=%s WHERE name=%s"data=(21,"张三")cursor.execute(sql,data)conn.commit()

3.删除数据

sql="DELETE FROM user WHERE name=%s"data=("张三",)cursor.execute(sql,data)conn.commit()

四、带异常处理（推荐）

生产环境必须用try-except-finally保证连接安全关闭，避免资源泄漏：

importpymysqlfrompymysqlimportErrordefconnect_mysql():try:# 建立连接conn=pymysql.connect(host="localhost",user="root",password="123456",database="test_db",charset="utf8mb4")ifconn.open:print("数据库连接成功！")cursor=conn.cursor()# 执行查询cursor.execute("SELECT VERSION()")version=cursor.fetchone()print("MySQL版本：",version)exceptErrorase:print("数据库连接失败：",e)finally:# 无论是否出错，都关闭连接if'conn'inlocals()andconn.open:cursor.close()conn.close()print("数据库连接已关闭")if__name__=="__main__":connect_mysql()

参数说明

五、MySQL 官方驱动

使用 MySQL 官方原生驱动，安装mysql-connector-python：

pipinstallmysql-connector-python

连接代码：

importmysql.connector conn=mysql.connector.connect(host="localhost",user="root",password="123456",database="test_db")

MySQL 数据库连接池

一、安装依赖

pipinstalldbutils pymysql

二、代码演示

importpymysqlfromdbutils.pooled_dbimportPooledDBclassMySQLPool:def__init__(self):self.pool=PooledDB(creator=pymysql,# 使用pymysql作为连接驱动maxconnections=10,# 连接池最大连接数mincached=2,# 初始化时至少创建的空闲连接maxcached=5,# 最大空闲连接maxshared=3,# 最大共享连接blocking=True,# 无连接时是否阻塞maxusage=None,# 单个连接最大复用次数setsession=[],# 执行SQL前的会话设置ping=0,# 自动检查连接可用性# 数据库配置host="127.0.0.1",port=3306,user="root",password="你的密码",database="test_db",charset="utf8mb4")# 获取连接defget_conn(self):returnself.pool.connection()# 查询（单行/多行）defquery(self,sql,args=None,one=False):conn=self.get_conn()cur=conn.cursor(pymysql.cursors.DictCursor)# 返回字典格式try:cur.execute(sql,args)ifone:returncur.fetchone()returncur.fetchall()finally:cur.close()conn.close()# 归还连接池，不是真关闭# 增删改（自动commit）defexecute(self,sql,args=None):conn=self.get_conn()cur=conn.cursor()try:cur.execute(sql,args)conn.commit()returncur.rowcountexceptExceptionase:conn.rollback()raiseefinally:cur.close()conn.close()# 全局单例连接池mysql_pool=MySQLPool()

三、使用示例

# 查询一条user=mysql_pool.query("SELECT * FROM user WHERE id=%s",(1,),one=True)print(user)# 查询所有users=mysql_pool.query("SELECT * FROM user")print(users)# 插入rows=mysql_pool.execute("INSERT INTO user(name,age) VALUES(%s,%s)",("张三",20))print("影响行数:",rows)