python 操作 MySQL
目录对于Python操作MySQL主要使用两种方式：
- 原生模块 pymsql
- ORM框架 SQLAchemy
pymsql是Python中操作MySQL的模块，其使用方法和MySQLdb几乎相同。
1 pymsql
1.1 下载安装
pip3 install pymysql
1.2 使用操作
1.2.1 执行SQL
import pymysql
import pymysql
# 创建连接
conn = pymysql.connect(host='localhost',port=3306,user='root',password='toor',db='careydb')
# 创建游标
cursor = conn.cursor()
# 执行SQL,并返回受影响行数
effict_row = cursor.execute("select * from stuffs")
print(effict_row)
print("<----获取单行数据---->")
print(cursor.fetchone())
print("<----获取多行数据---->")
print(cursor.fetchmany(3))
print("<----获取所有数据---->")
# fetchall()从当前指针所在位置开始读取
print(cursor.fetchall())
注：在fetch数据时按照顺序进行，可以使用cursor.scroll(num,mode)来移动游标位置，如：
cursor.scroll(1,mode='relative') # 相对当前位置移动
cursor.scroll(2,mode='absolute') # 相对绝对位置移动
1.2.2 插入表内容
insert_data = [
("user1", 23, "2019-10-01", 'M'),
("user2", 53, "2001-03-01", 'M'),
("user3", 14, "2009-05-01", 'M'),
]
cursor.executemany("insert into stuffs(name,times,register_date,sex) values(%s,%s,%s,%s)", insert_data)
# 提交，不然无法保存新建或者修改的数据
conn.commit()
cursor.close()
conn.close()
3. fetch数据类型
默认获取的数据是元组类型,若获取字典类型的数据：
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
r = cursor.execute("call p1()")
result = cursor.fetchone()
conn.commit()
cursor.close()
conn.close()
1.2.3 fetch数据类型
默认获取的数据是元组类型,若获取字典类型的数据：
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
r = cursor.execute("call p1()")
result = cursor.fetchone()
conn.commit()
cursor.close()
conn.close()
2 SQLAchemy
SQLAlchemy是Python编程语言下的一款ORM(object relational mapping)框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
优点：
- 隐藏了数据访问细节，“封闭”的通用数据库交互，ORM的核心。他使得我们的通用数据库交互变得简单易行，并且完全不用考虑SQL语句。快速开发。
- ORM使我们构造固化数据结构变得简单易行。
缺点：
- 自动化意味着映射和关联管理，代价是牺牲性能
架构图
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...]
2.1 安装
pip3 install SQLAlchemy
2.2 基本使用
2.2.1 Filter与Filter_by
filter用类名.属性名，赋值用"=="，不支持组合查询，只能连续调用filter来变相实现。
filter_by直接用属性名，赋值用"="，支持组合查询。
如,要实现组合查询：
filter：
data = sess.query(Table_user).filter(Table_user.name == 'carey').filter(Table_user.password == 'password').all()
或者：
data = sess.query(Table_user).filter(Table_user.name == 'carey' and Table_user.password == 'password').all()
filter_by：
data = sess.query(Table_user).filter_by(name='carey', password='password').all()
2.2.2 创建一个表
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
# 连接数据库
engine = create_engine("mysql+pymysql://root:toor@localhost/careydb",
encoding='utf-8', echo=True)
# 生成ORM基类
orm_base = declarative_base()
class Table_user(orm_base):
__tablename__ = 'user' # 表名
id = Column(Integer, primary_key=True)
name = Column(String(32))
password = Column(String(64))
# 创建表结构
orm_base.metadata.create_all(engine)
# 源代码中的_tablename_格式写错; 解决办法：把报错信息中的__tablename__复制过去就对了。
2.2.3 增
# 创建与数据库的会话Session calss，注意这里返回的是class不是实例
Session_class = sessionmaker(bind=engine)
# 实例化session
Session = Session_class()
# 生成要创建的数据对象
user_obj = Table_user(name='carey', password='123456')
# 此时还没有创建对象，打印ID结果为none
print(user_obj)
# 将要创建的数据对象添加到session中，此时仍然没有创建
Session.add(user_obj)
print(user_obj.name, user_obj.id)
# 统一提交，创建数据
Session.commit()
Session.close()
2.2.4 删
print(Session.query(Table_user).filter_by().all())
Session.query(Table_user).filter_by(name='user1').delete()
Session.commit()
# [1 name:user1, 2 name:fan, 3 name:carey, 4 name:carey]
print(Session.query(Table_user).filter_by().all())
# [2 name:fan, 3 name:carey, 4 name:carey]
2.2.5 改
data = Session.query(Table_user).filter(Table_user.id>0).filter(Table_user.id<2).first()
data.name = 'user1'
data.password = 'abc123'
Session.commit()
2.2.6 查
data = Session.query(Table_user).filter_by(name='carey')
print(data)
# filter_by查询的结果是一组数据，因此打印出来的是一个查询语句
print(data[1].name, data[1].password)
print("<----------all()&reper----------->")
data = Session.query(Table_user).filter_by(name='carey').all()
print(data)
# 打印出来的是所查询的对象，all()将所有数据取成一个列表，如“删”实例的结果。将所有需要在class Table_user()下
# 加入__repr__() 方法打印出表属性,
print(data[1].name, data[1].password)
print(data)
print("<----------first()----------->")
data = Session.query(Table_user).filter_by(name='carey').first()
print(data)
print("<----------使用条件筛选----------->")
data = Session.query(Table_user).filter(Table_user.id>0).filter(Table_user.id<2).all()
print(data)
# 运行结果
SELECT user.id AS user_id, user.name AS user_name, user.password AS user_password
FROM user
WHERE user.name = %(name_1)s
carey 654321
<----------all()&reper----------->
[1 name:carey, 3 name:carey]
carey 654321
[1 name:carey, 3 name:carey]
<----------first()----------->
1 name:carey
<----------使用条件筛选----------->
[1 name:carey]
2.2.7 回滚
chang_user= Session.query(Table_user).filter_by(name='user1').first()
chang_user.name = 'carey'
fake_user = Table_user(name='user2', password='abc123')
Session.add(fake_user)
print(Session.query(Table_user).filter(Table_user.name.in_(['carey', 'user2'])).all())
# 在结果中可以看到表中已经存在新添加的数据
Session.rollback()
print(Session.query(Table_user).filter(Table_user.name.in_(['carey', 'user2'])).all())
# 回滚后，新加数据消失。
2.2.8 统计分组
# 统计
print(Session.query(Table_user).filter(Table_user.name.like('ca%')).count())
# 分组
print(Session.query(Table_user.name,func.count(Table_user.name)).group_by(Table_user.name).all())
2.2.9 连表
class Table_stuffs(orm_base):
__tablename__ = 'stuffs'
id = Column(Integer,primary_key=True)
name = Column(String(32),nullable=False)
register_date = Column(Date,nullable=False)
sex = Column(String(32), nullable=False)
def __repr__(self):
return "%s name:%s" % (self.id,self.name)
print(Session.query(Table_user,Table_stuffs).filter(Table_user.id == Table_stuffs.id).all())
# print(session.query(Table_user).join(Table_stuffs).all())
# print(session.query(Table_user).join(Table_stuffs, isouter=True).all())
# 上面两条命令，因为Table_user与Table_stuffs没有进行外键关联，因此无法建立join关系
Session.commit()
Session.close()
2.2.10 单外键关联
# 学生与上课记录数据相关联，可以根据学生的名字查找其上课信息；
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import Column, Integer, String, Date, ForeignKey
engine = create_engine("mysql+pymysql://root:toor@localhost/careydb")
base = declarative_base()
class Table_students(base):
__tablename__ = 'students'
id = Column(Integer, primary_key=True)
name = Column(String(32),nullable=False)
register_date = Column(Date,nullable=True)
def __repr__(self):
return "<%s name: %s>" % (self.id,self.name)
class Table_StudyRecord(base):
__tablename__ = 'StudyRecord'
id = Column(Integer,primary_key=True)
day = Column(Integer,nullable=False)
status = Column(String(32),nullable=False)
stu_id = Column(Integer,ForeignKey(Table_students.id))
student = relationship(Table_students, backref="my_study_record")
# 这个nb，允许在students表里通过backref字段"my_study_record"反向查出所有它在studyrecord表里的关联项
def __repr__(self):
return "<%s day: %s status: %s>" % (self.id,self.day,self.status)
base.metadata.create_all(engine)
Session_class = sessionmaker(bind=engine)
Session = Session_class()
# student1 = Table_students(name='user1', register_date='2013-01-02')
# student2 = Table_students(name='user2', register_date='2014-10-23')
# student3 = Table_students(name='user3', register_date='2012-05-22')
# student4 = Table_students(name='user4', register_date='2011-01-11')
#
# study_record1 = Table_StudyRecord(day=1,status='yes', stu_id=1)
# study_record2 = Table_StudyRecord(day=3,status='no', stu_id=2)
# study_record3 = Table_StudyRecord(day=2,status='yes', stu_id=3)
# study_record4 = Table_StudyRecord(day=1,status='yes', stu_id=2)
#
# Session.add_all([student1,student2,student3,student4,study_record1,study_record2,study_record3,study_record4])
#
# Session.add_all([study_record1,study_record2,study_record3,study_record4])
student_obj = Session.query(Table_students).filter(Table_students.name=='user2').first()
print(student_obj)
print(student_obj.my_study_record)
Session.commit()
Session.close()
2.2.11 多外键关联
# 创建消费者与付款地址，邮寄地址之间的关系，每个消费者的付款地址与邮寄地址可能不同。
- orm_MulFK.py 创建表结构关系
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:toor@localhost/careydb", encoding='utf-8')
orm_base = declarative_base()
class Table_address(orm_base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True)
street = Column(String(64), nullable=False)
city = Column(String(32), nullable=False)
state = Column(String(32), nullable=False)
def __repr__(self):
return "<id:%s street: %s city: %s state: %s >" %(self.id,self.street,self.city,self.state)
class Table_customer(orm_base):
__tablename__ = 'customer'
id = Column(Integer,primary_key=True)
name = Column(String(32), nullable=False)
billing_addr_id = Column(Integer,ForeignKey(Table_address.id))
shipping_addr_id = Column(Integer,ForeignKey(Table_address.id))
billing_addr = relationship(Table_address, foreign_keys=[billing_addr_id])
shipping_addr = relationship(Table_address, foreign_keys=[shipping_addr_id])
def __repr__(self):
return "<id:%s name: %s >" %(self.id,self.name)
orm_base.metadata.create_all(engine)
- orm_MulFK_api.py 创建数据接口
from Day12 import orm_MulFK
from sqlalchemy.orm import sessionmaker
Session_class = sessionmaker(bind=orm_MulFK.engine)
Session = Session_class()
# addr1 = orm_MulFK.Table_address(street='Hubinnan', city='XiaMen', state='FJ')
# addr2 = orm_MulFK.Table_address(street='Fanshen', city='ShenZhen', state='GD')
# addr3 = orm_MulFK.Table_address(street='Guchenxilu', city='LiJiang', state='YN')
#
# cus1 = orm_MulFK.Table_customer(name='carey',billing_addr=addr1, shipping_addr=addr2)
# cus2 = orm_MulFK.Table_customer(name='fan',billing_addr=addr3, shipping_addr=addr3)
#
# Session.add_all([addr1,addr2,addr3,cus1,cus2])
obj = Session.query(orm_MulFK.Table_customer).filter(orm_MulFK.Table_customer.name=='carey').first()
print(obj.name, obj.billing_addr, obj.shipping_addr)
Session.commit()
Session.close()
2.2.12多对多外键关系
# 处理解决多对多关系，每个书有多个作者，每个作者可以写多本书；
- orm_M2MFK.py
from sqlalchemy import Table,Column, Integer, String, Date, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:toor@localhost/careydb", encoding='utf-8')
orm_base = declarative_base()
book_m2m_author = Table('book_m2m_author', orm_base.metadata,
Column('book_id', Integer, ForeignKey('books.id')),
Column('author_id', Integer, ForeignKey('authors.id')),
)
class Table_authors(orm_base):
__tablename__ = 'authors'
id = Column(Integer, primary_key=True)
name = Column(String(32))
def __repr__(self):
return self.name
class Table_books(orm_base):
__tablename__ = 'books'
id = Column(Integer, primary_key=True)
name = Column(String(32), nullable=False)
pub_date = Column(Date)
authors = relationship(Table_authors, secondary=book_m2m_author, backref='books')
def __repr__(self):
return self.name
orm_base.metadata.create_all(engine)
- orm_M2MFK_api.py
from Day12 import orm_M2MFK
from sqlalchemy.orm import sessionmaker
Session_class = sessionmaker(bind=orm_M2MFK.engine)
Session = Session_class()
# book1 = orm_M2MFK.Table_books(name='TCP/IP', pub_date='2002-01-03')
# book2 = orm_M2MFK.Table_books(name='python', pub_date='2007-10-19')
# book3 = orm_M2MFK.Table_books(name='CCIE', pub_date='2008-04-27')
#
# author1 = orm_M2MFK.Table_authors(name='carey')
# author2 = orm_M2MFK.Table_authors(name='fan')
# author3 = orm_M2MFK.Table_authors(name='God')
#
# book1.authors = [author1,author3]
# book3.authors = [author1,author2,author3]
#
# Session.add_all([book1,book2,book3,author1,author2,author3])
author_obj = Session.query(orm_M2MFK.Table_authors).filter(orm_M2MFK.Table_authors.name=='carey').first()
print(author_obj)
# 直接删除作者
# Session.delete(author_obj)
book_obj = Session.query(orm_M2MFK.Table_books).filter(orm_M2MFK.Table_books.id==2).first()
print(book_obj.name ,book_obj.authors)
# 通过书删除作者
# book_obj.authors.remove(author_obj)
# 处理中文，mysql8*上测试无需此操作
book3 = orm_M2MFK.Table_books(name='渗透测试', pub_date='2008-04-27')
Session.add(book3)
# engine = create_engine("mysql+pymysql://root:toor@localhost/careydb?charset=utf8", encoding='utf-8')
Session.commit()
Session.close()