Python 连接 MySQL 数据库

Python 连接 MySQL 数据库在 python 中连接 MySQL 数据库最常用、最稳定的方法是使用 PyMySQL 库。一、准备工作1.安装 PyMySQL 库pipinstallpymysql2.前提条件本地 / 远程已安装 MySQL 服务已创建数据库拥有 MySQL 账号密码确保 MySQL 端口默认3306未被防火墙拦截二、连接 MySQL# 1. 导入pymysql库importpymysql# 2. 创建数据库连接核心代码connpymysql.connect(hostlocalhost,# 数据库地址本地用localhost/127.0.0.1userroot,# MySQL用户名password123456,# MySQL密码databasetest_db,# 要连接的数据库名port3306,# MySQL默认端口charsetutf8mb4# 字符集支持emoji推荐)# 3. 创建游标对象用于执行SQL语句cursorconn.cursor()# 4. 执行SQL查询语句sqlSELECT * FROM user# 替换为你的表名cursor.execute(sql)# 5. 获取查询结果# fetchone()获取1条数据 | fetchall()获取所有数据resultscursor.fetchall()forrowinresults:print(row)# 打印每一行数据# 6. 关闭游标和连接必须释放资源cursor.close()conn.close()三、核心操作查询无需提交事务增 / 删 / 改必须执行conn.commit()提交事务否则数据不会生效1.插入数据importpymysql connpymysql.connect(hostlocalhost,userroot,password123456,databasetest_db,charsetutf8mb4)cursorconn.cursor()# 插入SQL占位符 %s 防止SQL注入推荐写法sqlINSERT 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.更新数据sqlUPDATE user SET age%s WHERE name%sdata(21,张三)cursor.execute(sql,data)conn.commit()3.删除数据sqlDELETE FROM user WHERE name%sdata(张三,)cursor.execute(sql,data)conn.commit()四、带异常处理推荐生产环境必须用try-except-finally保证连接安全关闭避免资源泄漏importpymysqlfrompymysqlimportErrordefconnect_mysql():try:# 建立连接connpymysql.connect(hostlocalhost,userroot,password123456,databasetest_db,charsetutf8mb4)ifconn.open:print(数据库连接成功)cursorconn.cursor()# 执行查询cursor.execute(SELECT VERSION())versioncursor.fetchone()print(MySQL版本,version)exceptErrorase:print(数据库连接失败,e)finally:# 无论是否出错都关闭连接ifconninlocals()andconn.open:cursor.close()conn.close()print(数据库连接已关闭)if__name____main__:connect_mysql()参数说明参数说明默认值host数据库地址localhostuser数据库用户名无password数据库密码无database要连接的数据库名无port端口号3306charset字符集推荐 utf8mb4utf8autocommit自动提交事务不推荐False五、MySQL 官方驱动使用 MySQL 官方原生驱动安装mysql-connector-pythonpipinstallmysql-connector-python连接代码importmysql.connector connmysql.connector.connect(hostlocalhost,userroot,password123456,databasetest_db)MySQL 数据库连接池一、安装依赖pipinstalldbutils pymysql二、代码演示importpymysqlfromdbutils.pooled_dbimportPooledDBclassMySQLPool:def__init__(self):self.poolPooledDB(creatorpymysql,# 使用pymysql作为连接驱动maxconnections10,# 连接池最大连接数mincached2,# 初始化时至少创建的空闲连接maxcached5,# 最大空闲连接maxshared3,# 最大共享连接blockingTrue,# 无连接时是否阻塞maxusageNone,# 单个连接最大复用次数setsession[],# 执行SQL前的会话设置ping0,# 自动检查连接可用性# 数据库配置host127.0.0.1,port3306,userroot,password你的密码,databasetest_db,charsetutf8mb4)# 获取连接defget_conn(self):returnself.pool.connection()# 查询单行/多行defquery(self,sql,argsNone,oneFalse):connself.get_conn()curconn.cursor(pymysql.cursors.DictCursor)# 返回字典格式try:cur.execute(sql,args)ifone:returncur.fetchone()returncur.fetchall()finally:cur.close()conn.close()# 归还连接池不是真关闭# 增删改自动commitdefexecute(self,sql,argsNone):connself.get_conn()curconn.cursor()try:cur.execute(sql,args)conn.commit()returncur.rowcountexceptExceptionase:conn.rollback()raiseefinally:cur.close()conn.close()# 全局单例连接池mysql_poolMySQLPool()三、使用示例# 查询一条usermysql_pool.query(SELECT * FROM user WHERE id%s,(1,),oneTrue)print(user)# 查询所有usersmysql_pool.query(SELECT * FROM user)print(users)# 插入rowsmysql_pool.execute(INSERT INTO user(name,age) VALUES(%s,%s),(张三,20))print(影响行数:,rows)