devbox
小舞很执着
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

python操作mysql_python mysql

作者：小舞很执着 | 2024-08-14 19:21:39

赞

踩

python mysql

一、python操作mysql简介

关于mysql安装及基本知识可以查看mysql专栏文章

https://blog.csdn.net/qq_34491508/category_11590629.html

python中可以使用 mysql-connector 来连接使用 MySQL， mysql-connector 是 MySQL 官方提供的驱动器，也可以使用PyMySQL 操作

关于mysql-connector的使用可以参考

Python MySQL – mysql-connector 驱动 | 菜鸟教程 (runoob.com)

无论通过何种方式去连接MySQL，本质上发送的 指令 都是相同的，只是连接的方式和操作形式不同而已。

这篇主要介绍开发中常用的PyMySQL 使用

PyMySQL 是在 Python3.x 版本中用于连接 MySQL 服务器的一个库，Python2 中则使用 mysqldb。

PyMySQL 遵循 Python 数据库 API v2.0 规范，并包含了 pure-Python MySQL 客户端库。

在使用 PyMySQL 之前，我们需要确保 PyMySQL 已安装。如果还未安装，我们可以使用以下命令安装最新版的 PyMySQL：

pip3 install PyMySQL

二、pymysql管理数据库

使用pymysql完成对数据库的增删改查

  1. import pymysql
  2.  
  3. # 连接MySQL（socket）
  4. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', charset="utf8")
  5. cursor = conn.cursor()
  6.  
  7. # 1. 查看数据库
  8. # 发送指令
  9. cursor.execute("show databases")
  10. # 获取指令的结果
  11. result = cursor.fetchall()
  12. print(result) # (('information_schema',), ('mysql',), ('performance_schema',), ('sys',))
  13.  
  14. # 2. 创建数据库（新增、删除、修改）
  15. # 发送指令
  16. cursor.execute("create database db3 default charset utf8 collate utf8_general_ci")
  17. conn.commit()
  18.  
  19. # 3. 查看数据库
  20. # 发送指令
  21. cursor.execute("show databases")
  22. # 获取指令的结果
  23. result = cursor.fetchall()
  24. print(result) # (('information_schema',), ('db3',), ('mysql',), ('performance_schema',), ('sys',))
  25.  
  26. # 4. 删除数据库
  27. # 发送指令
  28. cursor.execute("drop database db3")
  29. conn.commit()
  30.  
  31. # 3. 查看数据库
  32. # 发送指令
  33. cursor.execute("show databases")
  34. # 获取指令的结果
  35. result = cursor.fetchall()
  36. print(result) # (('information_schema',), ('mysql',), ('performance_schema',), ('sys',))
  37.  
  38. # 5. 进入数据库，查看表
  39. # 发送指令
  40. cursor.execute("use mysql")
  41. cursor.execute("show tables")
  42. result = cursor.fetchall()
  43. print(result) # (('columns_priv',), ('db',), ('engine_cost',), ('event',), ('func',), ('general_log',),....
  44.  
  45. # 关闭连接
  46. cursor.close()
  47. conn.close()

三、pymysql管理数据表

  1. import pymysql
  2.  
  3. # 连接MySQL
  4. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', charset="utf8")
  5. cursor = conn.cursor()
  6.  
  7. # 1. 创建数据库
  8. """
  9. cursor.execute("create database db4 default charset utf8 collate utf8_general_ci")
  10. conn.commit()
  11. """
  12.  
  13. # 2. 进入数据库、查看数据表
  14. """
  15. cursor.execute("use db4")
  16. cursor.execute("show tables")
  17. result = cursor.fetchall()
  18. print(result)
  19. """
  20.  
  21. # 3. 进入数据库创建表
  22. cursor.execute("use db4")
  23. sql = """
  24. create table L4(
  25.     id int not null primary key auto_increment,
  26.     title varchar(128),
  27.     content text,
  28.     ctime datetime
  29. )default charset=utf8;
  30. """
  31. cursor.execute(sql)
  32. conn.commit()
  33.  
  34. # 4. 查看数据库中的表
  35. """
  36. cursor.execute("show tables")
  37. result = cursor.fetchall()
  38. print(result)
  39. """
  40.  
  41. # 5. 其他 drop table... 略过
  42.  
  43.  
  44. # 关闭连接
  45. cursor.close()
  46. conn.close()

四、pymysql管理数据行

1、数据的增删改查

  1. import pymysql
  2.  
  3. # 连接MySQL，自动执行 use userdb; -- 进入数据库
  4. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', charset="utf8", db='userdb')
  5. cursor = conn.cursor()
  6.  
  7.  
  8. # 1.新增（需commit）
  9. """
  10. cursor.execute("insert into tb1(name,password) values('李四','123123')")
  11. conn.commit()
  12. """
  13.  
  14. # 2.删除（需commit）
  15. """
  16. cursor.execute("delete from tb1 where id=1")
  17. conn.commit()
  18. """
  19.  
  20. # 3.修改（需commit)
  21. """
  22. cursor.execute("update tb1 set name='xx' where id=1")
  23. conn.commit()
  24. """
  25.  
  26. # 4.查询
  27. """
  28. cursor.execute("select * from tb where id>10")
  29. data = cursor.fetchone() # cursor.fetchall()
  30. print(data)
  31. """
  32.  
  33. # 关闭连接
  34. cursor.close()
  35. conn.close()

2、案例：实现 注册、登录功能

  1. import pymysql
  2.  
  3.  
  4. def register():
  5.     print("用户注册")
  6.  
  7.     user = input("请输入用户名：") # alex
  8.     password = input("请输入密码：") # sb
  9.  
  10.     # 连接指定数据
  11.     conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', charset="utf8", db="usersdb")
  12.     cursor = conn.cursor()
  13.  
  14.     # 执行SQL语句（有SQL注入风险，稍后讲解）
  15.     # sql = 'insert into users(name,password)values("alex","sb")'
  16.     sql = 'insert into users(name,password) values("{}","{}")'.format(user, password)
  17.     
  18.     cursor.execute(sql)
  19.     conn.commit()
  20.  
  21.     # 关闭数据库连接
  22.     cursor.close()
  23.     conn.close()
  24.  
  25.     print("注册成功，用户名：{},密码:{}".format(user, password))
  26.  
  27.  
  28. def login():
  29.     print("用户登录")
  30.  
  31.     user = input("请输入用户名：")
  32.     password = input("请输入密码：")
  33.  
  34.     # 连接指定数据
  35.     conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', charset="utf8", db="usersdb")
  36.     cursor = conn.cursor()
  37.  
  38.     # 执行SQL语句（有SQL注入风险，稍后讲解）
  39.     # sql = select * from users where name='xxx' and password='123'
  40.     sql = "select * from users where name='{}' and password='{}'".format(user, password)
  41.     cursor.execute(sql)
  42.     
  43.     result = cursor.fetchone() # 去向mysql获取结果
  44.     # None
  45.     # (1,wupeiqi,123)
  46.     
  47.     
  48.     # 关闭数据库连接
  49.     cursor.close()
  50.     conn.close()
  51.  
  52.     if result:
  53.         print("登录成功", result)
  54.     else:
  55.         print("登录失败")
  56.  
  57.  
  58. def run():
  59.     choice = input("1.注册；2.登录")
  60.     if choice == '1':
  61.         register()
  62.     elif choice == '2':
  63.         login()
  64.     else:
  65.         print("输入错误")
  66.  
  67.  
  68. if __name__ == '__main__':
  69.     run()

3、关于SQL注入

  1. import pymysql
  2.  
  3. # 输入用户名和密码
  4. user = input("请输入用户名：") # ' or 1=1 -- 
  5. pwd = input("请输入密码：") # 123
  6.  
  7.  
  8. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', charset="utf8",db='usersdb')
  9. cursor = conn.cursor()
  10.  
  11. # 基于字符串格式化来 拼接SQL语句
  12. # sql = "select * from users where name='alex' and password='123'"
  13. # sql = "select * from users where name='' or 1=1 -- ' and password='123'"
  14. sql = "select * from users where name='{}' and password='{}'".format(user, pwd)
  15. cursor.execute(sql)
  16.  
  17. result = cursor.fetchone()
  18. print(result) # None，不是None
  19.  
  20. cursor.close()
  21. conn.close()

如果用户在输入user时，输入了： ' or 1=1 -- ，这样即使用户输入的密码不存在，也会可以通过验证。

为什么呢？

因为在SQL拼接时，拼接后的结果是：

select * from users where name='' or 1=1 -- ' and password='123'

注意：在MySQL中 -- 表示注释。

那么，在Python开发中 如何来避免SQL注入呢？

切记，SQL语句不要在使用python的字符串格式化，而是使用pymysql的execute方法。

  1. import pymysql
  2.  
  3. # 输入用户名和密码
  4. user = input("请输入用户名：")
  5. pwd = input("请输入密码：")
  6.  
  7. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='root123', charset="utf8", db='userdb')
  8.  
  9. cursor = conn.cursor()
  10.  
  11. cursor.execute("select * from users where name=%s and password=%s", [user, pwd])
  12. # 或
  13. # cursor.execute("select * from users where name=%(n1)s and password=%(n2)s", {"n1": user, 'n2': pwd})
  14.  
  15. result = cursor.fetchone()
  16. print(result)
  17.  
  18. cursor.close()
  19. conn.close()

五、工具类

上面pymysql对mysql的关键代码就是sql语句，其他都是公共的，所以可以封装成简单的数据库操作工具类

  1. import pymysql
  2.  
  3. # 连接地址相关可以从配置文件读取
  4. dataSource = {
  5.     'host': 'localhost',
  6.     'port': 3306,
  7.     'user': 'root',
  8.     'password': 'root',
  9.     'db': 'db_sys',
  10.     'charset': 'utf8'
  11. }
  12.  
  13.  
  14. def get_conn_cursor():
  15.     conn = pymysql.connect(host=dataSource['host'], port=dataSource['port'], user=dataSource['user'],
  16.                            password=dataSource['password'], charset=dataSource['charset'],
  17.                            db=dataSource['db'])
  18.     cursor = conn.cursor()
  19.     return conn, cursor
  20.  
  21.  
  22. def close_conn_cursor(*args):
  23.     for item in args:
  24.         item.close()
  25.  
  26.  
  27. def exec(sql, **kwargs):
  28.     conn, cursor = get_conn_cursor()
  29.     cursor.execute(sql, kwargs)
  30.     conn.commit()
  31.     close_conn_cursor(conn, cursor)
  32.  
  33.  
  34. def fetch_one(sql, **kwargs):
  35.     conn, cursor = get_conn_cursor()
  36.     cursor.execute(sql, kwargs)
  37.     result = cursor.fetchone()
  38.     close_conn_cursor(conn, cursor)
  39.     return result
  40.  
  41.  
  42. def fetch_all(sql, **kwargs):
  43.     conn, cursor = get_conn_cursor()
  44.     cursor.execute(sql, kwargs)
  45.     result = cursor.fetchall()
  46.     close_conn_cursor(conn, cursor)
  47.     return result
  48.  
  49.  
  50. if __name__ == '__main__':
  51.     users = fetch_all("select * from user")
  52.     print(users)  # ((1, 'admin', 'admin', None, None), (2, 'test', 'test', None, None))

六、数据库连接池

在操作数据库时需要使用数据库连接池。

  1. pip3 install pymysql
  2. pip3 install dbutils
  1. import threading
  2. import pymysql
  3. from dbutils.pooled_db import PooledDB
  4.  
  5. MYSQL_DB_POOL = PooledDB(
  6.     creator=pymysql,  # 使用链接数据库的模块
  7.     maxconnections=5,  # 连接池允许的最大连接数，0和None表示不限制连接数
  8.     mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
  9.     maxcached=3,  # 链接池中最多闲置的链接，0和None不限制
  10.     blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
  11.     setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
  12.     ping=0,
  13.     # ping MySQL服务端，检查是否服务可用。
  14.     # 如：0 = None = never, 1 = default = whenever it is requested, 
  15.     # 2 = when a cursor is created, 4 = when a query is executed, 7 = always
  16.     host='127.0.0.1',
  17.     port=3306,
  18.     user='root',
  19.     password='root123',
  20.     database='userdb',
  21.     charset='utf8'
  22. )
  23.  
  24.  
  25. def task():
  26.     # 去连接池获取一个连接
  27.     conn = MYSQL_DB_POOL.connection()
  28.     cursor = conn.cursor(pymysql.cursors.DictCursor)
  29.     
  30.     cursor.execute('select sleep(2)')
  31.     result = cursor.fetchall()
  32.     print(result)
  33.  
  34.     cursor.close()
  35.     # 将连接交换给连接池
  36.     conn.close()
  37.  
  38. def run():
  39.     for i in range(10):
  40.         t = threading.Thread(target=task)
  41.         t.start()
  42.  
  43.  
  44. if __name__ == '__main__':
  45.     run()

七、基于数据库连接池工具类

1、第一种：单例和方法工具类

  1. # db.py
  2. import pymysql
  3. from dbutils.pooled_db import PooledDB
  4.  
  5.  
  6. class DBHelper(object):
  7.  
  8.     def __init__(self):
  9.         # TODO 此处配置，可以去配置文件中读取。
  10.         self.pool = PooledDB(
  11.             creator=pymysql,  # 使用链接数据库的模块
  12.             maxconnections=5,  # 连接池允许的最大连接数，0和None表示不限制连接数
  13.             mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
  14.             maxcached=3,  # 链接池中最多闲置的链接，0和None不限制
  15.             blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
  16.             setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
  17.             ping=0,
  18.             # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
  19.             host='127.0.0.1',
  20.             port=3306,
  21.             user='root',
  22.             password='root123',
  23.             database='userdb',
  24.             charset='utf8'
  25.         )
  26.  
  27.     def get_conn_cursor(self):
  28.         conn = self.pool.connection()
  29.         cursor = conn.cursor(pymysql.cursors.DictCursor)
  30.         return conn, cursor
  31.  
  32.     def close_conn_cursor(self, *args):
  33.         for item in args:
  34.             item.close()
  35.  
  36.     def exec(self, sql, **kwargs):
  37.         conn, cursor = self.get_conn_cursor()
  38.  
  39.         cursor.execute(sql, kwargs)
  40.         conn.commit()
  41.  
  42.         self.close_conn_cursor(conn, cursor)
  43.  
  44.     def fetch_one(self, sql, **kwargs):
  45.         conn, cursor = self.get_conn_cursor()
  46.  
  47.         cursor.execute(sql, kwargs)
  48.         result = cursor.fetchone()
  49.  
  50.         self.close_conn_cursor(conn, cursor)
  51.         return result
  52.  
  53.     def fetch_all(self, sql, **kwargs):
  54.         conn, cursor = self.get_conn_cursor()
  55.  
  56.         cursor.execute(sql, kwargs)
  57.         result = cursor.fetchall()
  58.  
  59.         self.close_conn_cursor(conn, cursor)
  60.  
  61.         return result
  62.  
  63.  
  64. db = DBHelper()

测试

  1. from db import db
  2.  
  3. db.exec("insert into d1(name) values(%(name)s)", name="666")
  4.  
  5. ret = db.fetch_one("select * from d1")
  6. print(ret)
  7.  
  8. ret = db.fetch_one("select * from d1 where id=%(nid)s", nid=3)
  9. print(ret)
  10.  
  11. ret = db.fetch_all("select * from d1")
  12. print(ret)
  13.  
  14. ret = db.fetch_all("select * from d1 where id>%(nid)s", nid=2)
  15. print(ret)

 

2、上下文管理方式工具类（推荐）

如果想要让他也支持 with 上下文管理。

with 获取连接：
	执行SQL（执行完毕后，自动将连接交还给连接池）
  1. # db_context.py
  2. import threading
  3. import pymysql
  4. from dbutils.pooled_db import PooledDB
  5.  
  6. POOL = PooledDB(
  7.     creator=pymysql,  # 使用链接数据库的模块
  8.     maxconnections=5,  # 连接池允许的最大连接数，0和None表示不限制连接数
  9.     mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
  10.     maxcached=3,  # 链接池中最多闲置的链接，0和None不限制
  11.     blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
  12.     setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
  13.     ping=0,
  14.     host='127.0.0.1',
  15.     port=3306,
  16.     user='root',
  17.     password='root123',
  18.     database='userdb',
  19.     charset='utf8'
  20. )
  21.  
  22.  
  23. class Connect(object):
  24.     def __init__(self):
  25.         self.conn = conn = POOL.connection()
  26.         self.cursor = conn.cursor(pymysql.cursors.DictCursor)
  27.  
  28.     def __enter__(self):
  29.         return self
  30.  
  31.     def __exit__(self, exc_type, exc_val, exc_tb):
  32.         self.cursor.close()
  33.         self.conn.close()
  34.  
  35.     def exec(self, sql, **kwargs):
  36.         self.cursor.execute(sql, kwargs)
  37.         self.conn.commit()
  38.  
  39.     def fetch_one(self, sql, **kwargs):
  40.         self.cursor.execute(sql, kwargs)
  41.         result = self.cursor.fetchone()
  42.         return result
  43.  
  44.     def fetch_all(self, sql, **kwargs):
  45.         self.cursor.execute(sql, kwargs)
  46.         result = self.cursor.fetchall()
  47.         return result

测试

  1. from db_context import Connect
  2.  
  3. with Connect() as obj:
  4.     # print(obj.conn)
  5.     # print(obj.cursor)
  6.     ret = obj.fetch_one("select * from d1")
  7.     print(ret)
  8.  
  9.     ret = obj.fetch_one("select * from d1 where id=%(id)s", id=3)
  10.     print(ret)

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小舞很执着/article/detail/980475
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号