ORM框架之SQLAlchemy

SQLAlchemy是Python编程语言下的一款开源软件。提供了SQL工具包及对象关系映射（ORM：Object Relational Mappers）工具。SQLAlchemy“采用简单的Python语言，为高效和高性能的数据库访问设计，实现了完整的企业级持久模型”。SQLAlchemy的理念是，SQL数据库的量级和性能重要于对象集合；而对象集合的抽象又重要于表和行。因此，SQLAlchemy采用了类似于Java里Hibernate的数据映射模型，而不是其他ORM框架采用的Active Record模型。

1.单数据表

　　1.1单数据表创建

#pip install SQLAlchemy模块

#(1)创建对象基类------类似django中的Model
from sqlalchemy.ext.declarative import declarative_base
Base=declarative_base()

#(2)定义类对象---表
from sqlalchemy import Column,Integer,String
class Student(Base):#继承基类
    __tablename__='student'#表名
    id=Column(Integer,primary_key=True,autoincrement=True)#类属性---表字段
    name=Column(String(32),nullable=False,index=True,unique=True)


#(3)创建数据库引擎（依赖pymysql模块驱动）
from sqlalchemy.engine import create_engine
engine=create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/sqlalchemy?charset=utf8")#源码有配置说明：'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'


# (4)创建表
Base.metadata.create_all(engine)#创建基于Base基类的engine数据库引擎中的所有表
# Base.metadata.drop_all(engine)#删除基于Base基类的engine数据库引擎中的所有表

　　1.2单数据表的增删改查

　　1.2.1增 

from sqlalchemy.orm import sessionmaker
from create_table import Student,engine

#创建与数据库的连接
session=sessionmaker(engine)#初始化与数据库引擎的连接类型
db_session=session()#创建数据库连接



#增
#单个增加add
s=Student(name='yang')#实例化对象
db_session.add(s)#创建插入语句（只在当前的缓存中执行,但是id会被占用）
db_session.commit()#提交数据库执行同步更新指令
db_session.close()#使用完之后最好关闭连接


#批量增加add_all
s_list=[Student(name='zhang'),Student(name='li')]
db_session.add_all(s_list)
db_session.commit()
db_session.close()#使用完之后最好关闭连接

 

　　1.2.2查

from sqlalchemy.orm import sessionmaker
from create_table import Student,engine

#创建与数据库的连接
session=sessionmaker(engine)#初始化与数据库引擎的连接类型
db_session=session()#创建数据库连接

#查query
# （在执行其他操作之后务必关闭数据库连接，否则查询的是上一次连接建立时导入缓存的数据，导致数据不准确）
print(db_session.query(Student))#直接打印出原生SQL：SELECT student.id AS student_id, student.name AS student_name FROM student

#普通查询
res_all=db_session.query(Student).all()#查询所有
print(res_all)
for i in res_all:
    print(i.id,i.name)


res_first=db_session.query(Student).first()#查询所有结果的第一个
print(res_first)
print(res_first.id,res_first.name)

res_one=db_session.query(Student)[1]#查询结果中选取某一个
# res_one=db_session.query(Student).all()[1]
print(res_one)
print(res_one.id,res_one.name)


res_many=db_session.query(Student)[:2]#查询结果中的连续的某几个
# res_many=db_session.query(Student).all()[:2]
print(res_many)
for i in res_many:
    print(i.id,i.name)

# 带条件的查询
res=db_session.query(Student).filter(Student.id>1)
print(res)#直接输出了原生SQL语句
#SELECT student.id AS student_id, student.name AS student_name
# FROM student
# WHERE student.id > %(id_1)s

#filter中的算数运算符过滤
res=db_session.query(Student).filter(Student.id>1).all()
print(res,[(i.id,i.name) for i in res],sep='
')
res=db_session.query(Student).filter(Student.id==5).all()
print(res,[(i.id,i.name) for i in res],sep='
')
res=db_session.query(Student).filter(Student.id!=1).all()
print(res,[(i.id,i.name) for i in res],sep='
')


#filter中的并列条件过滤(条件之间用逗号隔开)
res=db_session.query(Student).filter(Student.id>1,Student.name!='yang').all()
print(res,[(i.id,i.name) for i in res],sep='
')

　　1.2.3改

from sqlalchemy.orm import sessionmaker
from create_table import Student,engine

#创建与数据库的连接
session=sessionmaker(engine)#初始化与数据库引擎的连接类型
db_session=session()#创建数据库连接

# 改(基于查询)
res=db_session.query(Student).filter(Student.name=='YANG').update({"name":'yang'})#单条修改，返回修改条数
db_session.commit()#务必要提交到数据库
print(res)
db_session.close()


res=db_session.query(Student).filter(Student.id>5).update({"name":'yang'})#更新多条，注意更新的该字段不能设置唯一，否则报错
db_session.commit()
print(res)#更新记录条数
db_session.close()

　　1.2.4删　　

from sqlalchemy.orm import sessionmaker
from create_table import Student,engine

#创建与数据库的连接
session=sessionmaker(engine)#初始化与数据库引擎的连接类型
db_session=session()#创建数据库连接

# 删（基于查询）
res=db_session.query(Student).filter(Student.id>8).delete()#返回删除的记录条数
db_session.commit()
print(res)
db_session.close()

　1.2.5 高级版查询操作 

# 高级版查询操作,厉害了哦
#老规矩
from create_table import Student,engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
db_session = Session()


# db_session.add_all([User(name='yang'),User(name='zhang'),User(name='wang')])
# db_session.commit()

# 查询数据表操作
# and or
from sqlalchemy.sql import and_ , or_,text
ret = db_session.query(Student).filter(and_(Student.id > 3, Student.name == 'yang')).all()
print(ret)
ret = db_session.query(Student).filter(or_(Student.id < 2, Student.name == 'yang')).all()
print(ret)

# # 查询所有数据
r1 = db_session.query(Student).all()

# # 查询数据 指定查询数据列 加入别名
r2 = db_session.query(Student.name.label('username'), Student.id).first()
print(r2.id,r2.username) # 3 wang

# 表达式筛选条件
r3 = db_session.query(Student).filter(Student.name == "yang").all()

# # 原生SQL筛选条件
r4 = db_session.query(Student).filter_by(name='yang').all()
r5 = db_session.query(Student).filter_by(name='yang').first()

# 字符串匹配方式筛选条件 并使用 order_by进行排序
r6 = db_session.query(Student).filter(text("id<:value and name=:name")).params(value=224, name='yang').order_by(Student.id).all()

# #原生SQL查询
r7 = db_session.query(Student).from_statement(text("SELECT * FROM User where name=:name")).params(name='yang').all()

# # 筛选查询列
# # query的时候我们不在使用User ORM对象,而是使用User.name来对内容进行选取
user_list = db_session.query(Student.name).all()
print(user_list)#[('wang',), ('yang',), ('zhang',)]
for row in user_list:
    print(row.name)
#
# # 别名映射  name as nick
user_list = db_session.query(Student.name.label("nick")).all()
print(user_list)
for row in user_list:
    print(row.nick) # 这里要写别名了
#
# # 筛选条件格式
user_list = db_session.query(Student).filter(Student.name == "yang").all()
user_list = db_session.query(Student).filter(Student.name == "yang").first()
user_list = db_session.query(Student).filter_by(name="yang").first()
for row in user_list:
    print(row.nick)

# 复杂查询
from sqlalchemy.sql import text
user_list = db_session.query(Student).filter(text("id<:value and name=:name")).params(value=3,name="DragonFire")

# 查询语句
from sqlalchemy.sql import text
user_list = db_session.query(Student).filter(text("select * from User id<:value and name=:name")).params(value=3,name="DragonFire")

# 排序 :
user_list = db_session.query(Student).order_by(Student.id).all()
user_list = db_session.query(Student).order_by(Student.id.desc()).all()
for row in user_list:
    print(row.name,row.id)

#其他查询条件
"""
ret = session.query(Student).filter_by(name='yang').all()
ret = session.query(Student).filter(Student.id > 1, Student.name == 'yang').all()
ret = session.query(Student).filter(Student.id.between(1, 3), Student.name == 'yang').all() # between 大于1小于3的
ret = session.query(Student).filter(Student.id.in_([1,3,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的
ret = session.query(Student).filter(~Student.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的
ret = session.query(Student).filter(Student.id.in_(session.query(Student.id).filter_by(name='yang'))).all() 子查询
from sqlalchemy import and_, or_
ret = session.query(Student).filter(and_(Student.id > 3, Student.name == 'yang')).all()
ret = session.query(Student).filter(or_(Student.id < 2, Student.name == 'yang')).all()
ret = session.query(Student).filter(
    or_(
        Student.id < 2,
        and_(User.name == 'eric', Student.id > 3),
        Student.extra != ""
    )).all()
# select * from Student where id<2 or (name="eric" and id>3) or extra != "" 

# 通配符
ret = db_session.query(Student).filter(Student.name.like('e%')).all()
ret = db_session.query(Student).filter(~Student.name.like('e%')).all()

# 限制
ret = db_session.query(Student)[1:2]

# 排序
ret = db_session.query(Student).order_by(Student.name.desc()).all()
ret = db_session.query(Student).order_by(Student.name.desc(), Student.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = db_session.query(Student).group_by(Student.extra).all()
ret = db_session.query(
    func.max(Student.id),
    func.sum(Student.id),
    func.min(Student.id)).group_by(Student.name).all()

ret = db_session.query(
    func.max(Student.id),
    func.sum(Student.id),
    func.min(Student.id)).group_by(Student.name).having(func.min(Student.id) >2).all()
"""

# 关闭连接
db_session.close()

 　　1.2.6高级版修改操作

#高级版更新操作
from create_table import Student,engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
db_session = Session()

#直接修改
db_session.query(Student).filter(Student.id > 0).update({"name" : "099"})

#在原有值基础上添加 - 1
db_session.query(Student).filter(Student.id > 0).update({Student.name: Student.name + "099"}, synchronize_session=False)

#在原有值基础上添加 - 2
db_session.query(Student).filter(Student.id > 0).update({"age": Student.age + 1}, synchronize_session="evaluate")
db_session.commit()
db_session.close()

2.一对多ForeignKey数据表(基于relationship)

 　　2.1ForeignKey一对多数据表及relationshipi关系创建

from sqlalchemy.ext.declarative import declarative_base
#(1)创建基类
Base=declarative_base()

from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship
#(2)创建对象，ForeignKey及relationship关系
class ClassTable(Base):
    __tablename__='classtable'
    id=Column(Integer,primary_key=True)
    name=Column(String(32),unique=False)

class Student(Base):
    __tablename__='student'
    id=Column(Integer,primary_key=True)
    name=Column(String(32),nullable=False)
    #创建foreignKey和relationship关系
    class_id=Column(Integer,ForeignKey('classtable.id')) # 关联字段,让class_id 与 classtable 的 id 进行关联,主外键关系(这里的ForeignKey一定要是表名.id不是对象名）
    stc=relationship("ClassTable",backref='cts')# 将student 与 ClassTable 创建关系 这个不是字段,只是关系,backref是反向关联的关键字


#(3)创建数据库引擎
from sqlalchemy import create_engine
engine=create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/sqlalchemy?charset=utf8")

#(4)基于数据库引擎创建集成基类的表
Base.metadata.create_all(engine)
# Base.metadata.drop_all(engine)

　　2.2一对多基于relationship的增查和改删操作

　　　　2.2.1一对多基于relationship的增

from sqlalchemy.orm import sessionmaker
from create_ForeignKey_table import Student,ClassTable,engine
#（1）建立数据库链接会话
session=sessionmaker(engine)
db_session=session()

#一对多的增

#普通的增

#先增加班级
db_session.add(ClassTable(name='三年一班'))
db_session.add_all([ClassTable(name='三年二班'),ClassTable(name='三年三班')])
db_session.commit()
#查询要添加到的班级
cla=db_session.query(ClassTable).filter(ClassTable.name=='三年一班').first()
#增加对应班级的学生
db_session.add(Student(name='yang',class_id=cla.id))
db_session.add_all([Student(name='zhang',class_id=cla.id),Student(name='li',class_id=cla.id)])
db_session.commit()
db_session.close()

#基于relationshi增加

#正向增加（foreiKey决定）
db_session.add(Student(name='he',stc=ClassTable(name='二年一班')))
db_session.commit()


#反向增加
class_obj=ClassTable(name='二年二班')#实例化班级或者查询出一个班级
# class_obj=db_session.query(ClassTable).filter(ClassTable.name=='二年二班').first()

# 向 Student 数据表中添加 1条或多条数据 并将 1条或多条数据的class_id 写成 class_obj的id
class_obj.cts=[Student(name='wu'),Student(name='feng')]#添加一个也必须是列表

db_session.add(class_obj)
db_session.commit()
db_session.close()

一对多基于relationship的增

　　　　2.2.2一对多基于relationship的查

from sqlalchemy.orm import sessionmaker
from create_ForeignKey_table import Student,ClassTable,engine
#（1）建立数据库链接会话
session=sessionmaker(engine)
db_session=session()

#查(基于relationship)
#正向查询
stu_list=db_session.query(Student).all()
print([(stu.id,stu.name,stu.stc.name,stu.stc.id) for stu in stu_list])

#反向查询
cla_list=db_session.query(ClassTable).all()
# print([(cla.id,cla.name,[stu.name for stu in cla.cts])for cla in cla_list])
for cla in cla_list:
    for stu in cla.cts:#通过relationship设置的backref反向关联到学生对象
        print(cla.id,cla.name,stu.name)

一对多基于relationship的查

　　　　2.2.3一对多的改　

from sqlalchemy.orm import sessionmaker
from create_ForeignKey_table import Student,ClassTable,engine
#（1）建立数据库链接会话
session=sessionmaker(engine)
db_session=session()

#通过班级修改学生信息
class_obj=db_session.query(ClassTable).filter(ClassTable.name=='二年二班').first()
db_session.query(Student).filter(Student.class_id==class_obj.id).update({'name':'22'})
db_session.commit()
db_session.close()


#通过学生修改班级信息
stu_obj=db_session.query(Student).filter(Student.name=='22').first()
db_session.query(ClassTable).filter(ClassTable.id==stu_obj.class_id).update({'name':'二年四班'})
db_session.commit()
db_session.close()

一对多的改

　　　　2.2.4一对多的删

from sqlalchemy.orm import sessionmaker
from create_ForeignKey_table import Student,ClassTable,engine
#（1）建立数据库链接会话
session=sessionmaker(engine)
db_session=session()

#删
class_obj=db_session.query(ClassTable).filter(ClassTable.name=='二年四班').first()
res=db_session.query(Student).filter(Student.class_id==class_obj.id).delete()
db_session.commit()
print(res)
db_session.close()

一对多的删

  3.多对多基于第三张的数据表

　　3.1多对多数据表relationship及第三张表ForeignKey创建

from sqlalchemy.ext.declarative import declarative_base
#（1）创建基类
Base=declarative_base()


from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship
#（2）创建对象和relationship关系
class Girl(Base):
    __tablename__='girl'
    id=Column(Integer,primary_key=True)
    name=Column(String(32),nullable=False)
    gtb=relationship('Boy',secondary='friendship', backref='btg')#创建多对多的relationship关系，注意第三张表的桥梁关系


class Boy(Base):
    __tablename__='boy'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)

#多对多关系必须创建第三张关联表
class Friendship(Base):
    __tablename__='friendship'
    id = Column(Integer, primary_key=True)
    girl_id=Column(Integer,ForeignKey('girl.id'))#创建于girl表的外键关联
    boy_id = Column(Integer, ForeignKey('boy.id'))#创建于boy表的外键关联

#（3）创建数据库引擎
from sqlalchemy import create_engine
engine=create_engine('mysql+pymysql://root:123456@127.0.0.1:3306/sqlalchemy?charset=utf8')

#（4）基于数据库引擎创建继承Base基类的表
Base.metadata.create_all(engine)

　　3.2多对多基于relationship的增查

　　　　3.2.1多对多基于relationship的增

from sqlalchemy.orm import sessionmaker
from create_MTM_table import engine,Girl,Boy
#(1)创建数据库连接
session=sessionmaker(engine)
db_session=session()


#多对多表的增
#MTM基于relationship的正向增
db_session.add(Girl(name='g1',gtb=[Boy(name='b1'),Boy(name='b11')]))
db_session.commit()

#MTM基于relationship的反向增
boy_obj=Boy(name='b2')
boy_obj.btg=[Girl(name='g2'),Girl(name='g22')]
db_session.add(boy_obj)
db_session.commit()

db_session.close()

　　　　3.2.1多对多基于relationship的查

from sqlalchemy.orm import sessionmaker
from create_MTM_table import engine,Girl,Boy
#(1)创建数据库连接
session=sessionmaker(engine)
db_session=session()

#多对多表的查
#通过girl查所有的boy
girl_list=db_session.query(Girl).all()
for girl in girl_list:
    for boy in girl.gtb:
        print(girl.name,"----------",boy.name)
#通过boy查询所有的girl
boy_list=db_session.query(Boy).all()
for boy in boy_list:
    for girl in boy.btg:
        print(boy.name, "----------", girl.name)