Python操作MySQL
Posted 樵夫
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Python操作MySQL相关的知识，希望对你有一定的参考价值。
本篇对于Python操作MySQL主要使用两种方式：
- 原生模块 PyMySQL
- ORM框架 SQLAchemy
一 PyMySQL
PyMySQL是Python3中操作MySQL的模块，其使用方法和Python2中MySQLdb几乎相同。
Django中也可以使用PyMySQL连接MySQL数据库。
1.下载安装
pip3 install pymysql
2.连接数据库
注意事项
在进行本文以下内容之前需要注意：
- 你有一个MySQL数据库，并且已经启动。
- 你有可以连接该数据库的用户名和密码
- 你有一个有权限操作的database
基本使用
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() # 定义要执行的SQL语句 sql = """ CREATE TABLE USER1 ( id INT auto_increment PRIMARY KEY , name CHAR(10) NOT NULL UNIQUE, age TINYINT NOT NULL )ENGINE=innodb DEFAULT CHARSET=utf8; """ # 执行SQL语句 cursor.execute(sql) # 关闭光标对象 cursor.close() # 关闭数据库连接 conn.close()
默认获取的数据是元祖类型，可以设置游标返回字典格式数据：
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句并且将结果作为字典返回的游标 cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) # 定义要执行的SQL语句 sql = """ CREATE TABLE USER1 ( id INT auto_increment PRIMARY KEY , name CHAR(10) NOT NULL UNIQUE, age TINYINT NOT NULL )ENGINE=innodb DEFAULT CHARSET=utf8; """ # 执行SQL语句 cursor.execute(sql) # 关闭光标对象 cursor.close() # 关闭数据库连接 conn.close()
注意：
charset=“utf8”，编码不要写成"utf-8"
防止SQL注入
#存在用户名注入 当用户输入用户名存在时，在密码栏填入 存在的用户名\' -- 就可以直接登陆，利用的是传统程序中的字符串拼接漏洞使用单引号结束SQL语句，--注释掉后面的密码验证语句 #不需要用户名和密码直接注入 当用户在用户栏填上 随便的用户名\' or 1==1 -- 用单引号结束SQL语句，使用或判断语句正确，注释后面的其他SQL语句，返回的是true，直接进入数据库 #PyMySQL中解决SQL注入方式 1.使用原生的字符串拼接时对注入的语句进行过滤处理然后再验证 2.不使用原生拼接方式，使用cursor.execute(sql,[value1,value2])或者cursor.executemany(sql,[(value1,value2),(value1,value2),(value1,value2)])进行传值做拼接，下面增删改查会具体说明
3.增删改查操作
增
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() sql = "INSERT INTO USER1(name, age) VALUES (%s, %s);" username = "李思" age = 18 # 执行SQL语句 cursor.execute(sql, [username, age]) # 提交事务 conn.commit() cursor.close() conn.close()
插入数据失败回滚
在执行增删改操作时，如果不想提交前面的操作，可以使用 rollback() 回滚取消操作。
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() sql = "INSERT INTO USER1(name, age) VALUES (%s, %s);" username = "李思" age = 18 try: # 执行SQL语句 cursor.execute(sql, [username, age]) # 提交事务 conn.commit() except Exception as e: # 有异常，回滚事务 conn.rollback() cursor.close() conn.close()
获取插入数据的ID(关联操作时会用到)
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() sql = "INSERT INTO USER1(name, age) VALUES (%s, %s);" username = "李思" age = 18 try: # 执行SQL语句 cursor.execute(sql, [username, age]) # 提交事务 conn.commit() # 提交之后，获取刚插入的数据的ID last_id = cursor.lastrowid except Exception as e: # 有异常，回滚事务 conn.rollback() cursor.close() conn.close()
批量执行
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() sql = "INSERT INTO USER1(name, age) VALUES (%s, %s);" data = [("李思", 18), ("樵夫", 20), ("静静", 21)] try: # 批量执行多条插入SQL语句 cursor.executemany(sql, data) # 提交事务 conn.commit() except Exception as e: # 有异常，回滚事务 conn.rollback() cursor.close() conn.close()
删
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() sql = "DELETE FROM USER1 WHERE id=%s;" try: cursor.execute(sql, [4]) # 提交事务 conn.commit() except Exception as e: # 有异常，回滚事务 conn.rollback() cursor.close() conn.close()
改
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() # 修改数据的SQL语句 sql = "UPDATE USER1 SET age=%s WHERE name=%s;" username = "李思" age = 80 try: # 执行SQL语句 cursor.execute(sql, [age, username]) # 提交事务 conn.commit() except Exception as e: # 有异常，回滚事务 conn.rollback() cursor.close() conn.close()
查
查询单条数据
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() # 查询数据的SQL语句 sql = "SELECT id,name,age from USER1 WHERE id=1;" # 执行SQL语句 cursor.execute(sql) # 获取单条查询数据 ret = cursor.fetchone() cursor.close() conn.close() # 打印下查询结果 print(ret)
查询多条数据
# 导入pymysql模块 import pymysql # 连接database conn = pymysql.connect(host=“你的数据库地址”, user=“用户名”,password=“密码”,database=“数据库名”,charset=“utf8”) # 得到一个可以执行SQL语句的光标对象 cursor = conn.cursor() # 查询数据的SQL语句 sql = "SELECT id,name,age from USER1;" # 执行SQL语句 cursor.execute(sql) # 获取多条查询数据 ret = cursor.fetchall() cursor.close() conn.close() # 打印下查询结果 print(ret)
4.进阶用法
注：在fetch数据时按照顺序进行，可以使用cursor.scroll(num,mode)来移动游标位置 # 可以获取指定数量的数据 cursor.fetchmany(3) # 光标按绝对位置移动1 cursor.scroll(1, mode="absolute") # 光标按照相对位置(当前位置)移动1 cursor.scroll(1, mode="relative")
# 执行SQL，并返回收影响行数
effect_row = cursor.execute("update hosts set host = \'1.1.1.2\'")
# 执行SQL，并返回受影响行数
#effect_row = cursor.execute("update hosts set host = \'1.1.1.2\' where nid > %s", (1,))
# 执行SQL，并返回受影响行数
#effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
参考表结构： 用户类型（角色role） id user_type 1 员工 2 主管 用户信息（userinfo） id name pwd 1 李思 123 2 樵夫 123 权限（permission） id permission 1 订单管理 2 bug管理 用户类型&权限（role_per） id role_id permisson_id 1 1 2 2 2 1 功能： # 登陆、注册、找回密码 # 用户管理 # 用户类型 # 权限管理 # 分配权限 特别的：程序仅一个可执行文件
二 ORM框架之SQLAchemy
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
ORM框架类型：
- DB first: 手动创建数据库以及表 -> ORM框架 -> 自动生成类
- code first： 手动创建类、和数据库 -> ORM框架 -> 以及表 （SQLAchemy）
1.安装：
pip3 install SQLAlchemy
SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：
MySQL-Python
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html
2、内部处理
使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。
#!/usr/bin/env python # -*- coding:utf-8 -*- from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5) #指定数据库类型以及数据库对应的第三方插件，数据库用户幸喜，max_overflow最大连接数量 # 执行SQL（cur相当于连接池中一个连接） # cur = engine.execute( # "INSERT INTO hosts (host, color_id) VALUES (\'1.1.1.22\', 3)" # ) # 新插入行自增ID # cur.lastrowid # 执行SQL # cur = engine.execute( # "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[(\'1.1.1.22\', 3),(\'1.1.1.221\', 3),] # ) # 执行SQL # cur = engine.execute( # "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)", # host=\'1.1.1.99\', color_id=3 # ) # 执行SQL # cur = engine.execute(\'select * from hosts\') # 获取第一行数据（一次一次的读到内存里） # cur.fetchone() # 获取第n行数据 # cur.fetchmany(3) # 获取所有数据（一下读到内存中）
3、ORM功能使用
使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有组件对数据进行操作。根据类创建对象，对象转换成SQL，执行SQL。
1、创建表
#!/usr/bin/env python # -*- coding:utf-8 -*- from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5) Base = declarative_base() # 创建单表 class Users(Base):
#创建表名称 __tablename__ = \'users\' id = Column(Integer, primary_key=True) name = Column(String(32)) extra = Column(String(16)) __table_args__ = ( UniqueConstraint(\'id\', \'name\', name=\'uix_id_name\'), Index(\'ix_id_name\', \'name\', \'extra\'), ) # 一对多 class Favor(Base): __tablename__ = \'favor\' nid = Column(Integer, primary_key=True) caption = Column(String(50), default=\'red\', unique=True) class Person(Base): __tablename__ = \'person\' nid = Column(Integer, primary_key=True) name = Column(String(32), index=True, nullable=True) favor_id = Column(Integer, ForeignKey("favor.nid")) # 多对多 class Group(Base): __tablename__ = \'group\' id = Column(Integer, primary_key=True) name = Column(String(64), unique=True, nullable=False) port = Column(Integer, default=22) class Server(Base): __tablename__ = \'server\' id = Column(Integer, primary_key=True, autoincrement=True) hostname = Column(String(64), unique=True, nullable=False) class ServerToGroup(Base): __tablename__ = \'servertogroup\' nid = Column(Integer, primary_key=True, autoincrement=True) server_id = Column(Integer, ForeignKey(\'server.id\')) group_id = Column(Integer, ForeignKey(\'group.id\')) #创建连接 def init_db(): Base.metadata.create_all(engine) #删除连接 def drop_db(): Base.metadata.drop_all(engine)
注：设置外检的另一种方式 ForeignKeyConstraint([\'other_id\'], [\'othertable.other_id\'])
2、操作表
#!/usr/bin/env python # -*- coding:utf-8 -*- from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5) Base = declarative_base() # 创建单表 class Users(Base): __tablename__ = \'users\' id = Column(Integer, primary_key=True) name = Column(String(32)) extra = Column(String(16)) __table_args__ = ( UniqueConstraint(\'id\', \'name\', name=\'uix_id_name\'), Index(\'ix_id_name\', \'name\', \'extra\'), ) def __repr__(self): return "%s-%s" %(self.id, self.name) # 一对多 class Favor(Base): __tablename__ = \'favor\' nid = Column(Integer, primary_key=True) caption = Column(String(50), default=\'red\', unique=True) def __repr__(self): return "%s-%s" %(self.nid, self.caption) class Person(Base): __tablename__ = \'person\' nid = Column(Integer, primary_key=True) name = Column(String(32), index=True, nullable=True) favor_id = Column(Integer, ForeignKey("favor.nid")) # 与生成表结构无关，仅用于查询方便 favor = relationship("Favor", backref=\'pers\') # 多对多 class ServerToGroup(Base): __tablename__ = \'servertogroup\' nid = Column(Integer, primary_key=True, autoincrement=True) server_id = Column(Integer, ForeignKey(\'server.id\')) group_id = Column(Integer, ForeignKey(\'group.id\')) group = relationship("Group", backref=\'s2g\') server = relationship("Server", backref=\'s2g\') class Group(Base): __tablename__ = \'group\' id = Column(Integer, primary_key=True) name = Column(String(64), unique=True, nullable=False) port = Column(Integer, default=22) # group = relationship(\'Group\',secondary=ServerToGroup,backref=\'host_list\') class Server(Base): __tablename__ = \'server\' id = Column(Integer, primary_key=True, autoincrement=True) hostname = Column(String(64), unique=True, nullable=False) def init_db(): Base.metadata.create_all(engine) def drop_db(): Base.metadata.drop_all(engine) Session = sessionmaker(bind=engine) session = Session()
基础操作
#增 obj = Users(name="alex0", extra=\'sb\') session.add(obj) session.add_all([ Users(name="alex1", extra=\'sb\'), Users(name="alex2", extra=\'sb\'), ]) session.commit() #删 session.query(Users).filter(Users.id > 2).delete() session.commit() #改 session.query(Users).filter(Users.id > 2).update({"name" : "099"}) session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False) session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate") session.commit() #查 ret = session.query(Users).all() ret = session.query(Users.name, Users.extra).all() ret = session.query(Users).filter_by(name=\'alex\').all() ret = session.query(Users).filter_by(name=\'alex\').first() ret = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name=\'fred\').order_by(User.id).all() ret = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name=\'ed\').all()
其他操作
# 条件 ret = session.query(Users).filter_by(name=\'alex\').all() ret = session.query(Users).filter(Users.id > 1, Users.name == \'eric\').all() ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == \'eric\').all() ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all() ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name=\'eric\'))).all() from sqlalchemy import and_, or_ ret = session.query(Users).filter(and_(Users.id > 3, Users.name == \'eric\')).all() ret = session.query(Users).filter(or_(Users.id < 2, Users.name == \'eric\')).all() ret = session.query(Users).filter( or_( Users.id < 2, and_(Users.name == \'eric\', Users.id > 3), Users.extra != "" )).all() # 通配符 ret = session.query(Users).filter(Users.name.like(\'e%\')).all() ret = session.query(Users).filter(~Users.name.like(\'e%\')).all() # 限制 ret = session.query(Users)[1:2] # 排序 ret = session.query(Users).order_by(Users.name.desc()).all() ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all() # 分组 from sqlalchemy.sql import func ret = session.query(Users).group_by(Users.extra).all() ret = session.query( func.max(Users.id), func.sum(Users.id), func.min(Users.id)).group_by(Users.name).all() ret = session.query( func.max(Users.id), func.sum(Users.id), func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all() # 连表 ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all() ret = session.query(Person).join(Favor).all() ret = session.query(Person).join(Favor, isouter=True).all() # 组合 q1 = session.query(Users.name).filter(Users.id > 2) q2 = session.query(Favor.caption).filter(Favor.nid < 2) ret = q1.union(q2).all() q1 = session.query(Users.name).filter(Users.id > 2) q2 = session.query(Favor.caption).filter(Favor.nid < 2) ret = q1.union_all(q2).all()
以上是关于Python操作MySQL的主要内容，如果未能解决你的问题，请参考以下文章