orm_sqlalchemy总结

sqlalchemy使用总结

1 sqlalchemy ORM基础操作

官方文档：https://docs.sqlalchemy.org/en/13/orm/tutorial.html

创建连接 - 创建基类 - 创建实体类 - 创建表 - 创建session

import logging
import pymysql
from pymysql.cursors import DictCursor
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy import text
from sqlalchemy.orm import declarative_base
from sqlalchemy import String, Integer, Date, Column
from sqlalchemy.orm import sessionmakerlogging.basicConfig(level=logging.INFO)
ip = "127.0.0.1"
port = 9998
addr = (ip, port)kwargs = {"host": ip, "password": "cli*963.", "user": "root", "database": "test"}# pymysql操作数据库
try:with pymysql.connect(**kwargs) as conn:with conn.cursor(DictCursor) as cursor:cursor.execute("select * from tee where id=%s", args=(10,))logging.info(cursor.fetchall())
except Exception as err:logging.error(err)# 实体类的基类
Base = declarative_base()# 实体类
class Student(Base):  # 表映射为类，行映射为类的实例，字段映射为属性__tablename__ = "student"  # 约定的每一个实体类必须定义此属性id = Column(Integer, primary_key=True)  # 字段映射为属性name = Column(String(20))age = Column(Integer, nullable=True)def __repr__(self):return "<{} id={},name={},age={}>".format(self.__class__.__name__, self.id, self.name, self.age)# 引擎，管理连接池
conn_str = "mysql+pymysql://{user}:{password}@{hostname}/{database}".format(user="root", password="cli*963.", hostname=ip, database="test")  # conn_str连接字符串,约定俗成名字，最好都用这个名字。
engine = create_engine(conn_str, echo=True)  # echo=True,便于在调试过程中查看生成了哪些sql语句
with engine.connect() as conn:result = conn.execute(text("select * from tee where id=10"))print(result.fetchall())# 创建表、删除表
Base.metadata.create_all(engine)
Base.metadata.drop_all(engine)# 创建Session
Session = sessionmaker(bind=engine)  # Session类
session = Session()  # Session实例instance, session对象线程不安全的，所以每个线程最好基于全局的Session类实例化一个

---

2 sqlalchemy CRUD操作

sqlalchemy的CRUD操作。

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import String, Integer, Date, Column
from sqlalchemy.orm import sessionmakerconn_str = "mysql+pymysql://{user}:{password}@{hostname}/{database}".format(user="root", password="cli*963.", hostname=ip, database="test")
engine = create_engine(conn_str, echo=True)
Base = declarative_base()
Session = sessionmaker(bind=engine)
session = Session()class Student(Base):  # 表映射为类，行映射为类的实例，字段映射为属性__tablename__ = "student"  # 约定的每一个实体类必须定义此属性id = Column(Integer, primary_key=True)  # 字段映射为属性name = Column(String(20))age = Column(Integer, nullable=True)def __repr__(self):return "<{} id={},name={},age={}>".format(self.__class__.__name__, self.id, self.name, self.age)# 1 增：创建记录的实例并增加到表中
try:tom = Student(name="tom")tom.age = 12session.add(tom)session.add_all([Student(name="Jerry", age=12),Student(name="Lily", age=33),Student(name="Xiao", age=21)])  # pending状态session.commit()
except Exception as err:print(err)session.rollback()  # 回滚
finally:# session.close()pass# 一条记录的增加、修改状态变化分析
try:lim = Student(name="lim")lim.age = 12session.add(lim)session.commit()session.add(lim)  # 注意这里，两次add，第一次add因为没有lim这一条记录，为insert语句；第二次add为update，因为存在lim这条记录了、有主键了，就对比状态有没有变化，没有任何变化，所以不做任何修改；如果对lim的属性做修改，则会有update语句。session.commit()
except Exception as err:print(err)session.rollback()
finally:# session.close()pass# 2 基础查询
print(session.query(Student).filter_by(name="tom").first())
print(session.query(Student).filter(Student.id == 1))
for instance in session.query(Student).order_by(Student.age):print(instance.name, instance.age)# 3 修改
# 修改过程：方法一、先增加后修改（上述两次add）；方法二、先查询后修改（不先查后改的话，认为是insert）
# 查询：使用get方法：get方法通过主键查询
try:student = session.query(Student).get(1)student.age = 23session.add(student)  # 修改update：需要先查询、再修改。 add相当于在session这里注册一下，pending状态，待session提交，有任何异常则回滚session.commit()
except Exception as err:print(err)session.rollback()
finally:# session.close()pass# 4 删除
# 删除过程：先查询后拿到主键，然后删除
try:student = session.query(Student).get(1)session.delete(student)session.commit()
except Exception as err:print(err)session.rollback()
finally:# session.close()passif __name__ == '__main__':print(sqlalchemy.__version__)

---

3 sqlalchemy 实体的状态分析

状态：每一个实体都有一个状态属性_sa_instance_state，其类型是sqlalchemy.orm.state.InstanceState，可以使用sqlalchemy.inspect()函数查看实体状态。
常见的状态值有transient、pending、persistent、deleted、detached。

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

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, inspect
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.state import InstanceStatehost = "127.0.0.1"
user = "root"
password = "cli*963."
database = "test"def show(entity):ins = inspect(entity)print("transient={}, pending={}, persistent={}, deleted={}, detached={}".format(ins.transient, ins.pending, ins.persistent, ins.deleted, ins.detached))conn_str = "mysql+pymysql://{user}:{pwd}@{hostname}/{database}".format(user=user, pwd=password, hostname=host, database=database)
engine = create_engine(conn_str, echo=True)
Base = declarative_base()class Employee(Base):__tablename__ = "employee"id = Column(Integer, primary_key=True)first_name = Column(String(20))second_name = Column(String(20))age = Column(Integer)department = Column(String(30))def __repr__(self):return "<{}(id={}, dept={}, age={}, name={})>".format(self.__class__.__name__, self.id, self.department, self.age, self.first_name + " " + self.second_name)Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()# 状态分析
emp = Employee(first_name="Tom", second_name="shit", age=19, department="product")
show(emp)
session.add(emp)
show(emp)
session.commit()
show(emp)
emp.age = 29
show(emp)
session.commit()
show(emp)if __name__ == '__main__':print(sqlalchemy.__version__)print(Employee.__table__)

• 新建一个实体，状态是transient临时的。

• 一旦add()后从transient变成pending状态。成功commit()后从pending变成persistent状态。成功查询返回的实体对象，也是persistent状态。

• persistent状态的实体，修改依然是persistent状态。

• persistent状态的实体，删除后，flush后但没有commit，就变成deteled状态，成功提交，变为detached状态，提交失败，还原到persistent状态。

• 删除、修改操作，需要对应一个真实的记录，所以要求实体对象是persistent状态。

4 sqlalchemy 复杂查询

4.1 各种查询方法汇总

import datetimeimport sqlalchemy
from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, inspect, Date, Enum
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.state import InstanceState
import enumhost = "127.0.0.1"
user = "root"
password = "cli*963."
database = "test"class MyEnum(enum.Enum):M = "M"F = "F"def show(empys):"""打印函数"""for ins in empys:print(ins, end='\n\n')Base = declarative_base()class Employee(Base):__tablename__ = "employee"emp_no = Column(Integer, primary_key=True)birth_date = Column(Date, nullable=False)first_name = Column(String(20), nullable=False)last_name = Column(String(20), nullable=False)gender = Column(Enum(MyEnum), nullable=False)age = Column(Integer)hire_date = Column(Date, nullable=False)def __repr__(self):return "<{}(id={}, name={}, age={}, gender={})>".format(self.__class__.__name__, self.emp_no, self.first_name + " " + self.last_name, self.age, self.gender)conn_str = "mysql+pymysql://{user}:{pwd}@{hostname}/{database}".format(user=user, pwd=password, hostname=host, database=database)
engine = create_engine(conn_str, echo=True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add_all([Employee(first_name="Tom", last_name="shit", age=19, birth_date=datetime.datetime.now(), gender=MyEnum.M, hire_date=datetime.datetime.now()),Employee(first_name="Lily", last_name="Gaga", age=18, birth_date=datetime.datetime.now(), gender=MyEnum.F, hire_date=datetime.datetime.now()),Employee(first_name="Jerry", last_name="Port", age=23, birth_date=datetime.datetime.now(), gender=MyEnum.M, hire_date=datetime.datetime.now()),Employee(first_name="Jack", last_name="Jos", age=21, birth_date=datetime.datetime.now(), gender=MyEnum.F, hire_date=datetime.datetime.now())
])
session.commit()# 1、简单查询
emps = session.query(Employee).filter(Employee.emp_no > 2)
show(emps)# 与或非
from sqlalchemy import or_, not_, and_
# and
emps = session.query(Employee).filter(and_(Employee.emp_no > 2, Employee.gender == MyEnum.M))
show(emps)
emps = session.query(Employee).filter(Employee.emp_no > 2).filter(Employee.gender == MyEnum.M)
show(emps)
show(session.query(Employee).filter((Employee.emp_no > 2) & (Employee.gender == MyEnum.M)))# or
emps = session.query(Employee).filter((Employee.emp_no > 5) | (Employee.emp_no < 2))
show(emps)
emps = session.query(Employee).filter(or_(Employee.emp_no > 5, Employee.emp_no < 2))
show(emps)# not
emps = session.query(Employee).filter(not_(Employee.emp_no > 5))
show(emps)
emps = session.query(Employee).filter(~(Employee.emp_no > 5))  # 一定注意加括号
show(emps)# in
emps_list = [1, 3, 4]
emps = session.query(Employee).filter(Employee.emp_no.in_(emps_list))
show(emps)
# not in
emps = session.query(Employee).filter(~Employee.emp_no.in_(emps_list))
show(emps)
# like
emps = session.query(Employee).filter(Employee.last_name.like("J%"))  # like可以忽略大小匹配
show(emps)# 2、 排序
# 升序
show(session.query(Employee).filter(Employee.emp_no > 3).order_by(Employee.emp_no))
show(session.query(Employee).filter(Employee.emp_no > 3).order_by(Employee.emp_no).asc())
# 降序
show(session.query(Employee).filter(Employee.emp_no > 3).order_by(Employee.emp_no).desc())
# 多列排序
show(session.query(Employee).filter(Employee.emp_no > 3).order_by(Employee.emp_no).order_by(Employee.last_name).desc())# 3、分页
show(session.query(Employee).limit(4))
show(session.query(Employee).limit(4).offset(6))# 3、消费者方法：消费者方法调用后，query对象转换成一个容器
# 总行数
emps = session.query(Employee)
print(len(list(emps)))  # 返回大量结果集，然后转换为list
print(emps.count())  # 聚合函数count(*)的查询。使用聚合函数取总行数效率更高更合理
# 取所有数据
print(emps.all())
# 取一行
print(emps.one())  # 返回一行，如果查询结果是多行会抛异常
print(emps.limit(1).one())
# 删除
session.query(Employee).filter(Employee.emp_no < 3).delete()
# session.commit()  # 提交则删除# 4、聚合、分组
# 聚合函数
from sqlalchemy import func
query = session.query(func.count(Employee.emp_no))
print(query.one())  # 只能有一行结果
print(query.scalar())  # 取one()返回元组的第一个元素
# max/min/avg
print(session.query(func.max(Employee.emp_no)).scalar())
print(session.query(func.min(Employee.emp_no)).scalar())
print(session.query(func.avg(Employee.emp_no)).scalar())
# 分组
print(session.query(func.count(Employee.emp_no)).group_by(Employee.gender).a11())if __name__ == '__main__':print(sqlalchemy.__version__)print(Employee.__table__)

4.2 关联查询

根据下列三张表，查询10010号员工的所在部门编号:

CREATE TABLE `employees` (`emp_no` int(11) NOT NULL,`birth_date` date NOT NULL,`first_name` varchar(14) NOT NULL,`last_name` varchar(16) NOT NULL,`gender` enum('M','F') NOT NULL,`hire_date` date NOT NULL,PRIMARY KEY (`emp_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;CREATE TABLE `departments` (`dept_no` char(4) NOT NULL,`dept_name` varchar(40) NOT NULL,PRIMARY KEY (`dept_no`),UNIQUE KEY `dept_name` (`dept_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;CREATE TABLE `dept_emp` (`emp_no` int(11) NOT NULL,`dept_no` char(4) NOT NULL,`from_date` date NOT NULL,`to_date` date NOT NULL,PRIMARY KEY (`emp_no`,`dept_no`),KEY `dept_no` (`dept_no`),CONSTRAINT `dept_emp_ibfk_1` FOREIGN KEY (`emp_no`) REFERENCES `employees` (`emp_no`) ON DELETE CASCADE,CONSTRAINT `dept_emp_ibfk_2` FOREIGN KEY (`dept_no`) REFERENCES `departments` (`dept_no`) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3;

从语句可以看出，员工、部门之间的关系是多对多的关系。先把这些表的Model类和字段属性建立起来。
使用ForeignKey(“employees.emp_no”, ondelete=“CASCADE”), 定义外键约束

需求： 查询10010号员工的所在部门编号:

from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, Date, Enum
from sqlalchemy.orm import sessionmaker
import enum
from sqlalchemy.orm import relationshiphost = "127.0.0.1"
user = "root"
password = "cli*963."
database = "test"class MyEnum(enum.Enum):M = "M"F = "F"def show(empys):"""打印函数"""for ins in empys:print(ins, end='\n\n')Base = declarative_base()
conn_str = "mysql+pymysql://{user}:{pwd}@{hostname}/{database}".format(user=user, pwd=password, hostname=host, database=database)
engine = create_engine(conn_str, echo=True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()class Employees(Base):__tablename__ = "employees"emp_no = Column(Integer, primary_key=True)birth_date = Column(Date, nullable=False)first_name = Column(String(20), nullable=False)last_name = Column(String(20), nullable=False)gender = Column(Enum(MyEnum), nullable=False)# age = Column("age", Integer)  # 第一个参数是字段名，如果属性名与字段名不一致，一定要指定字段名。hire_date = Column(Date, nullable=False)def __repr__(self):return "<{}(id={}, name={}, gender={})>".format(self.__class__.__name__, self.emp_no, self.first_name + " " + self.last_name, self.gender)class Departments(Base):__tablename__ = "departments"dept_no = Column(String(4), primary_key=True)dept_name = Column(String(40), nullable=False, unique=True)def __repr__(self):return "<{}(dept_no={}, dept_name={})>".format(self.__class__.__name__, self.dept_no, self.dept_name)class DeptEmp(Base):__tablename__ = "dept_emp"emp_no = Column(Integer, ForeignKey("employees.emp_no", ondelete="CASCADE"), primary_key=True)dept_no = Column(String(4), ForeignKey("departments.dept_no", ondelete="CASCADE"), primary_key=True)from_date = Column(Date, nullable=False)to_date = Column(Date, nullable=False)def __repr__(self):return "<{}(dept_no={}, emp_no={})>".format(self.__class__.__name__, self.dept_no, self.emp_no)# 需求：查询10010号员工的所在部门编号
# 1、使用隐式内连接
res = session.query(Employees, DeptEmp).filter(Employees.emp_no == DeptEmp.emp_no).filter(Employees.emp_no == 10010).all()
show(res)
# 查询结果
# (Employee no=10010 name=Duangkaew Piveteau gender=F, Dept_emp empno-10010 deptno=d004)
# (Employee no=10010 name=Duangkaew Piveteau gender=F, Dept_emp empno=10010 deptno=d006)
# 这种方式产生隐式连接语句：
# SELECT * FROM employees, dept_emp WHERE employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s;# 2、使用join
# 第一种写法：不推荐，由ORM来查找内连接，有风险
show(session.query(Employees).join(DeptEmp).filter(Employees.emp_no == 10010).all())
# 第二种写法：
show(session.query(Employees).join(DeptEmp, Employees.emp_no == DeptEmp.emp_no).filter(Employees.emp_no == 10010).all())# 这两种写法，返回都只有一行数据，为什么？
# 原因在于query(Employee）这个只能返回一个实体对象中去，为了解决这个问题，需要修改实体类Employee，增加属性用来存放部门信息
# sqlalchemy.orm.relationship(实体类名字符串)
class Employees1(Employees):dept_emps = relationship("DeptEmp")  # tells the ORM that the Address class itself should be linked to the User class,def __repr__(self):return "<{}(id={}, name={}, age={}, gender={}, no={})>".format(self.__class__.__name__, self.emp_no, self.first_name + " " + self.last_name, self.age, self.gender,self.dept_emps)results = session.query(Employees).join(DeptEmp, (Employees1.emp_no == DeptEmp.emp_no)&(Employees1.emp_no == 10010))
show(results.all())

• 第一种方法join(DeptEmp)中没有等值条件，会自动生成一个等值条件，如果后面有fiter，哪怕是filter(Employees.emp_no==DeptEmp.emp_no)，这个条件会在where中出现。第一种这种自动增加join的等值条件的方式不好，不要这么写

• 第二种方法在join中增加等值条件，阳止了自动的等值条件的生成。这种方式推荐

• 第三种方法就是第二种，这种方式也可以。