python之SQLAIchemy

2025-01-25 技术教程

一简介和安装1 简介1 简介

SQLALchemy 是一个第三方的ORM 框架，大量使用元编程实现其基本功能

相关参考文档:https://docs.sqlalchemy.org/en/13/

2 基本组成

一个连接池引擎中存放着连接池和方言(dialect)，dialect不同的数据库引擎对应不同的方言。

DBAPI通过API调用数据库的各种功能，如控制语句等。

2 安装1 安装

pip install sqlalchemy2 版本检测

pip show sqlalchemy二基本实践1 创建连接

数据库链接的事情，交给引擎处理

#!/usr/bin/poython3.6#conding:utf-8import pymysqlimport sqlalchemyfrom sqlalchemy import create_engine#创建存储引擎username='root'password='Admin@Root123'ip='192.168.1.200'port=3306databases='test'# echo =True 打印SQL执行过程，主要应用于调试模式engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)2 创建实体类

#!/usr/bin/poython3.6#conding:utf-8import pymysqlimport sqlalchemyfrom sqlalchemy import create_engine# 导入基类模块from sqlalchemy.ext.declarative import declarative_base# 导入实体类相关属性模块from sqlalchemy import Column, Integer, String#创建存储引擎username='root'password='Admin@Root123'ip='192.168.1.200'port=3306databases='test'engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)# 导入基类Base=declarative_base()# 创建实体类class Login(Base): __tablename__='login' id=Column(Integer,primary_key=True) # 设置id列为整形，并设置其为主键，整形时Inteager name=Column(String(48)) # 定义name名称，并定义其字段为charset类型和长度为48 loginname=Column(String(48)) password=Column(String(256)) def __repr__(self): return "<Login (id='%s',name='%s',loginname='%s',password='%s')>" %(self.id,self.name,self.loginname,self.password)3 实例化

#实例化L=Login(name='admin')L.age=304 创建表

可以使用SQLAIchemy 来创建，删除表
删除继承自Base的所有表
Base.metadata.drop_all(engine)
创建继承自Base的所有表
Base.metadata.create_all(engine)
生产环境很少这样创建表，都是系统上线的时候由脚本生成的，生产环境很少删除表，废弃都不能删除。

具体代码如下

#!/usr/bin/poython3.6#conding:utf-8import pymysqlimport sqlalchemyfrom sqlalchemy import create_engine# 导入基类模块from sqlalchemy.ext.declarative import declarative_base# 导入实体类相关属性模块from sqlalchemy import Column, Integer, String#创建存储引擎username='root'password='666666'ip='192.168.1.120'port=3306databases='test'engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)# 导入基类Base=declarative_base()# 创建实体类class Login(Base): __tablename__='login' id=Column(Integer,primary_key=True) # 设置id列为整形，并设置其为主键，整形时Inteager name=Column(String(48)) # 定义name名称，并定义其字段为charset类型和长度为48 loginname=Column(String(48)) password=Column(String(256)) def __repr__(self): return "<Login (id='%s',name='%s',loginname='%s',password='%s')>" %(self.id,self.name,self.loginname,self.password)#将对应实体类加入存储引擎。执行相关指令Base.metadata.create_all(engine) # 此处通过引擎提供的连接数据库的能力来创建默认字符集和相关检测方式，其被存储在Base元类中。

执行结果如下

数据库结果如下

5 创建会话

在一个会话中操作数据库，会话建立在连接上，连接被引擎管理

#创建会话Session = sessionmaker(bind=engine) #此处返回的是一个类，需要对其进行实例化相关操作，此处不会commit#对返回的类进行实例化# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了session=Session()6 CRUD 操作(增，删，改，查)1 增1 增加单行数据

结果如下

数据库结果如下

2 同时插入多行

add_all() 相关源码

def add_all(self, instances): """Add the given collection of instances to this ``Session``.""" if self._warn_on_events: self._flush_warning("Session.add_all()") for instance in instances: # 此处也是通过add的方式进行遍历加入的 self.add(instance, _warn=False)

代码如下

结果如下

数据库结果如下

3 多次提交commit是否提交多次问题

结果如下

结论：如上，多次提交时commit只执行一次。

4 修改更新数据并提交

执行代码结果显示如下

数据库结果如下

结论:
常规来说，上述代码应该是提交了两次，其并未进行相关的限制，但其显示结果却是修改了，而不是两次提交，此处便涉及到了状态变化，应该是进行了某种判断，此处add后状态未发生变化，因此只添加了一条

insert 只能返回你添加了几行，下面的情况会修改，其修改会使用update进行处理，更新的时候，一般需要给ID(主键)进行处理，如果没有，则insert后拿取对应的id进行相关的处理，及提交完成后若出现更改，则通过一定方式获取上次提交的主键，并进行相关的操作处理，当一次commit提交完成后，实例的无主键状态会变为有主键状态。

2 简单查询语句

使用query()方法，返回一个Query对象

查询相关源码

def query(self, *entities, **kwargs): """Return a new :class:`.Query` object corresponding to this :class:`.Session`.""" return self._query_cls(entities, self, **kwargs)

上述可知，其查询参数是可变参数和关键字参数，及支持多级查询

相关代码如下

结果如下

3 修改

结果如下

数据库结果如下

具体步骤如下
1 先获取到查询结果
2 进行相关的修改操作
3 最后提交

4 删除

结果如下

5 状态处理

上述产生一个异常，是未进行持久化的异常，每一个实体，都有状态属性_sa_instance_state，其类型是sqlchemy.orm.state.InstanceState,可以使用sqlalchemy.inspect(entity)函数查看状态

常见的状态值有 transient，pending，persistent，deleted，detached。

状态说明transient实体类尚未加入到session中，同时并没有保存到数据库中pendingtransient的实体被add()到session中，状态切换到pending，但是还没有flush到数据库中persistentsession中的实体对象对应着数据库中的真实记录，pending状态在提交成功后可以变成persistent状态，或者查询成功返回的实体也是persistent状态deleted实体被删除且已经flush但未commit完成，事务提交成功了，实体变成detached，事务失败，返回persistentdetached删除成功的实体进入此状态

解释
新建一个实体，状态是transient临时的
一旦add()后从transient状态变为pending状态
成功commit()后从pending变为persistent状体
成功查询返回的实体对象，也是persistent状态
及就是 persistent状态可以由两种形式组成(commit和 get成功)
persistent状态的实体，修改依然是persistent状态
persistent状态的实体，删除后，flus后但没有commit，就变成了deleted状态，成功提交，变成detached状态，失败，还原persistent状态
删除，修改操作，需要对应一个真实的记录，所以要求实体对象必须是persistent状态

状态处理实例

结果如下

7 复杂查询语句1基本查询1 简单条件查询语句

emps=session.query(Employess).filter(Employess.emp_no > 10025) # 工号大于10025 的显示2 in 语句

emps=session.query(Employess).filter(Employess.emp_no.in_(emplist))3 not in 语句

emps=session.query(Employess).filter(~Employess.emp_no.in_(emplist))4 消费者方法

消费者方法调用后，Query对象(可迭代)就转换成了一个容器

# 消费者方法 emps=session.query(Employess) # 查询所有 print (emps.all()) # 返回列表 print (emps.count()) # 聚合函数查询 print (len(list(emps))) # 返回列表元素数量，及返回值数量 emps=session.query(Employee).filter(Employee.emp_no==10025)5 LIKE

emps=session.query(Employess).filter(Employess.last_name.like('P%'))6 上述结果如下

#!/usr/bin/poython3.6#conding:utf-8import pymysqlimport sqlalchemyfrom sqlalchemy import create_engine# 导入基类模块from sqlalchemy.ext.declarative import declarative_base# 导入实体类相关属性模块from sqlalchemy import Column, Integer, String,Enum,DATE# 导入会话模块from sqlalchemy.orm import sessionmaker# 导入枚举类型import enum#创建存储引擎username='root'password='666666'ip='192.168.1.120'port=3306databases='test'engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))# 导入基类Base=declarative_base()# 创建要查询的表对应的实体类#表字段如下'''CREATE TABLE `employees` ( `emp_no` int(11) NOT NULL COMMENT '主键', `birth_date` date NOT NULL COMMENT '生日', `first_name` varchar(14) NOT NULL COMMENT '用户-姓', `last_name` varchar(16) NOT NULL COMMENT '用户-名', `gender` enum('M','F') NOT NULL COMMENT '性别', `hire_date` date NOT NULL COMMENT '入职时间', PRIMARY KEY (`emp_no`)) ENGINE=InnoDB DEFAULT CHARSET=utf8;'''# 创建枚举类class MyEnum(enum.Enum): M='M' F='F'# 创建对应实体类class Employess(Base): __tablename__='employees' emp_no=Column(Integer,nullable=False,primary_key=True) # 整形，非空，主键 birth_date=Column(DATE,nullable=False) # 日期类型，非空 first_name=Column(String(14),nullable=False) # 字符串类型，长度为14，非空 last_name=Column(String(14),nullable=False) # 字符串类型，长度为16，非空 gender=Column(Enum(MyEnum),nullable=False) # 枚举，非空 hire_date=Column(DATE,nullable=False) # 字符类型，非空 def __repr__(self): # 定义查询返回 return "< {} emp_no:{} name:{}> ".format(self.__class__.__name__,self.emp_no,"{} {}".format(self.emp_no,self.last_name)) __str__=__repr__#将对应实体类加入存储引擎。执行相关指令Base.metadata.create_all(engine)#创建会话Session = sessionmaker(bind=engine) #此处返回的是一个类，需要对其进行实例化相关操作#对返回的类进行实例化# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了session=Session()# 实例化并进行插入操作def show(emps): for x in emps: print (x) print ('------------',end='\n\n')try: # 简单查询语句 emps=session.query(Employess).filter(Employess.emp_no > 10025) # 工号大于10025 的显示 print ("简单查询语句") show(emps) # in 语句查询 emplist=[10001,10002,10025] emps=session.query(Employess).filter(Employess.emp_no.in_(emplist)) print ("in 语句查询") show(emps) # not in 语句查询 emps=session.query(Employess).filter(~Employess.emp_no.in_(emplist)) print ("not in 语句查询") show(emps) # 消费者方法 emps=session.query(Employess) # 查询所有 print ("消费者方法") print (emps.all()) # 返回列表 print (emps.count()) # 聚合函数查询 print (len(list(emps))) # 返回列表元素数量，及返回值数量 # like emps=session.query(Employess).filter(Employess.last_name.like('P%')) print ('Like方法') print (show(emps))except Exception as e: # 若抛出异常，则直接回滚 print ('异常如下：--------',e)finally: pass

结果如下

2 聚合，分组1 聚合函数

# 导入聚合函数from sqlalchemy import functry: # count命令 print (session.query(func.count(Employess.emp_no)).all()) print (session.query(func.count(Employess.emp_no)).one()) # 若使用one,则只能由一个结果，若有多个结果，则会报错 print (session.query(func.count(Employess.emp_no)).scalar()) #取回one()元组返回的第一个元素 # AVG 平均值 print (session.query(func.avg(Employess.emp_no)).all()) print (session.query(func.avg(Employess.emp_no)).scalar()) # min 最小值 print (session.query(func.min(Employess.emp_no)).all()) print (session.query(func.min(Employess.emp_no)).scalar()) # max 最大值 print (session.query(func.max(Employess.emp_no)).all()) print (session.query(func.max(Employess.emp_no)).scalar())except Exception as e: # 若抛出异常，则直接回滚 print ('异常如下：--------',e)finally: pass

结果如下

2 分组

print (session.query(Employess.gender,func.count(Employess.emp_no)).group_by(Employess.gender).all())

结果如下

3 逻辑与或非

# 导入与或非模块from sqlalchemy import or_,and_,not_ # AND 第一个和第二个都要成立 # 方式一and emps=session.query(Employess).filter(Employess.emp_no>10015).filter(Employess.emp_no<10018) show(emps) # 方式二 and emps=session.query(Employess).filter(and_(Employess.emp_no>10015,Employess.emp_no<10018)) show(emps) # 方式三 & emps=session.query(Employess).filter((Employess.emp_no>10015)&(Employess.emp_no<10018)) show(emps) # OR 或 # 方式一 emps=session.query(Employess).filter((Employess.emp_no>10025)|(Employess.emp_no<10005)) show(emps) # 方式二 emps=session.query(Employess).filter(or_(Employess.emp_no>10025),(Employess.emp_no>10005)) show(emps) #NOT # 方式一 emps=session.query(Employess).filter(not_(Employess.emp_no>10002)) show(emps) #方式二 emps=session.query(Employess).filter(~(Employess.emp_no>10002)) show(emps)

4 排序和分页

# 排序，默认是升序排列 # 升序排列 emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.emp_no) print ("升序排列") show(emps) emps = session.query(Employess).filter(Employess.emp_no > 10025).order_by(Employess.emp_no.asc()) show(emps) # 降序排列 emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.emp_no.desc()) print ("降序排列") show(emps) # 多列排序,first_name 为降序排列,emp_no为升序排列，默认只有两个first_name相同时，后面的才有意义 emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.first_name.desc()).order_by(Employess.emp_no) print ("多列排序") show(emps) # 分页 emps=session.query(Employess).limit(4) print ("分页") show(emps) emps=session.query(Employess).limit(4).offset(20) show(emps)

结果如下

5 查询删除

emps=session.query(Employess.emp_no).filter(Employess.emp_no>10025).delete() print ('查询结果为:',emps) # session.commit()

结果如下

查询到的东西就是想要的，直接删除，此处对其进行了优化，直接使用where语句进行处理，而没有经过select处理

6 关联查询1 需求

查询10010 员工所在的部门编号
使用隐式内连接

# 一般连接方式处理如下 results=session.query(Employess,Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all() show(results) # 隐式内连接第一种写法 results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==10010).all() show(results) # 隐式内连接第二种写法 results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all() print (results)

第一种写法最终翻译结果如下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date, dept_emp.emp_no AS dept_emp_emp_no, dept_emp.dept_no AS dept_emp_dept_no, dept_emp.from_date AS dept_emp_from_date, dept_emp.to_date AS dept_emp_to_date FROM employees, dept_emp WHERE employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s

第二种写法最终翻译结果如下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no WHERE employees.emp_no = %(emp_no_1)s

第三种写法最终翻译结果如下

第四种写法最终翻译结果如下

显示如下

这种写法，都返回一行数据，原因是query(Employee)这只能返回一个实体对象中去，为了解决这个问题，需要修改实体类Employee，增加属性用来存放部门信息
sqlalchemy.orm.relationship(实体类名字符串)

2 使用relationship来显示其他表的信息

#!/usr/bin/poython3.6#conding:utf-8import pymysqlimport sqlalchemyfrom sqlalchemy import create_engine# 导入基类模块from sqlalchemy.ext.declarative import declarative_base# 导入实体类相关属性模块from sqlalchemy import Column, Integer, String,Enum,DATE,ForeignKey# 导入会话模块from sqlalchemy.orm import sessionmaker,relationship# 导入枚举类型import enum#创建存储引擎username='root'password='666666'ip='192.168.1.120'port=3306databases='test'engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))# 导入基类Base=declarative_base()# 创建要查询的表对应的实体类class Dept_emp(Base): __tablename__='dept_emp' emp_no=Column(Integer,ForeignKey('employees.emp_no',ondelete='CASAED'),primary_key=True) # 整形，非空，主键,ForeignKey用于定义外键， dept_no=Column(String(4),ForeignKey('departments.dept_no',ondelete='CASCAED'),primary_key=True) from_date=Column(DATE,nullable=False) to_date=Column(DATE,nullable=False) def __repr__(self): return "< {} empno={} dept_no={}>".format(self.__class__.__name__,self.emp_no,self.dept_no)# 创建枚举类class MyEnum(enum.Enum): M='M' F='F'# 创建对应实体类class Employess(Base): __tablename__='employees' emp_no=Column(Integer,nullable=False,primary_key=True) # 整形，非空，主键 birth_date=Column(DATE,nullable=False) # 日期类型，非空 first_name=Column(String(14),nullable=False) # 字符串类型，长度为14，非空 last_name=Column(String(14),nullable=False) # 字符串类型，长度为16，非空 gender=Column(Enum(MyEnum),nullable=False) # 枚举，非空 hire_date=Column(DATE,nullable=False) # 字符类型，非空 # 增加属性 depts=relationship('Dept_emp') # 后面接一个类名，这几个的外键关系已经被记录了，在dept_mp中已经表达过了，生成语句的投影和employee有关，此处生成一个属性来存储部门之间的关系，如果相反，则无法记录上面的不存在的名字，是依靠Colum中的外键中进行定义处理这个东西的 def __repr__(self): # 定义查询返回 return "< {} emp_no:{} name:{} depts:{}> ".format(self.__class__.__name__,self.emp_no,"{} {}".format(self.first_name,self.last_name),self.depts) __str__=__repr__class Department(Base): __tablename__='departments' dept_no=Column(String(32),primary_key=True) dept_name=Column(String(40),nullable=False,unique=True) def __repr__(self): return "< {} no={} name={}>".format(self.__class__.__name__,self.dept_no,self.dept_name) __str__=__repr__#将对应实体类加入存储引擎。执行相关指令Base.metadata.create_all(engine)#创建会话Session = sessionmaker(bind=engine) #此处返回的是一个类，需要对其进行实例化相关操作#对返回的类进行实例化# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了session=Session()# 实例化并进行插入操作def show(emps): for x in emps: print (x) print ('------------',end='\n\n')try: # 一般连接方式处理如下 results=session.query(Employess,Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all() show(results) # 隐式内连接第一种写法，此处的query 写谁，最终其显示结果就是谁的表的列处理 results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==10010).all() show(results) # 隐式内连接第二种写法 results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all() show(results) # 隐式连接第三种想法 results=session.query(Employess).join(Dept_emp,(Employess.emp_no==Dept_emp.emp_no)&(Employess.emp_no==10010)).all() show(results)except Exception as e: # 若抛出异常，则直接回滚 print ('异常如下：--------',e)finally: pass

结果如下

3 总结 :

第二种写法中join(Dept_empo)中没有等值条件，会自动生成等值条件，如果后面有filter，哪怕是filter(Employess.emp_no==Dept_emp.emp_no),这个条件会在where中出现，第二种自动增加的join等值条件的方式不好，不建议这样写
第三种在join中增加等值条件，组织了自动的等值条件生成，这种方式是推荐的
第四种方式和第三种是相同的，也可以使用

8 总结