ORM
ORM，对象关系映射，对象和关系之间的映射，使用面向对象的方式来操作数据库。
关系模型和Python对象之间的映射
- table => class ，表映射为类
- row => object ，行映射为实例
- column => property ，字段映射为属性
SQLAlchemy
是一个ORM框架
安装 pip install sqlalchemy
文档
查看版本
import sqlalchemy
print(sqlalchemy.__version__)
开发
内部使用了 连接池
创建连接
交给引擎来做
格式为： dialect+driver://username:password@host:port/database
mysqldb的连接
格式 ： mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
代码 ： engine = sqlalchemy.create_engine("mysql+mysqldb://wayne:wayne@127.0.0.1:3306/magedu")
pymysql的连接
格式 ： mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
engine = sqlalchemy.create_engine("mysql+pymysql://wayne:wayne@127.0.0.1:3306/magedu")
engine = sqlalchemy.create_engine("mysql+pymysql://wayne:wayne@127.0.0.1:3306/magedu",echo=True)
echo=True ： 引擎是否打印执行的语句，调试的时候打开很方便。
懒连接： 创建引擎并不会马上连接数据库，直到让数据库执行任务时才连接。从连接池分配一个连接。
Declare a Mapping创建映射
创建基类
from sqlalchemy.ext.declarative import declarative_base# 创建基类，便于实体类继承。SQLAlchemy大量使用了元编程Base = declarative_base()
创建实体类
# 创建实体类class Student(Base): # 指定表名 __tablename__ = 'student' # 定义类属性对应字段 id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(64), nullable=False) age = Column(Integer) # 第一参数是字段名，如果和属性名不一致，一定要指定 # age = Column('age', Integer) def __repr__(self): return "{} id={} name={} age={}".format( self.__class__.__name__, self.id, self.name, self.age)
__tablename__ 指定表名
Column类指定对应的字段，必须指定
实例化
s = Student(name='tom')print(s.name)s.age = 20print(s.age)
创建表
可以使用SQLAlchemy来创建、删除表
# 删除继承自Base的所有表Base.metadata.drop_all(engine)# 创建继承自Base的所有表Base.metadata.create_all(engine)
metadata 记录的是继承自Base的表，
生产环境很少这样创建表，都是系统上线的时候由脚本生成。
生成环境很少删除表， 宁可废弃都不能删除。
创建会话session
在一个会话中操作数据库，会话建立在连接上，连接被引擎管理。
当第一次使用数据库时，从引擎维护的连接池中获取一个连接使用。
# 创建sessionSession = sessionmaker(bind=engine) # 返回类session = Session() # 实例化# 依然在第一次使用时连接数据库
session对象线程不安全。所以不同线程应该使用不用的session对象。
Session类和engine有一个就行了。
import sqlalchemyfrom sqlalchemy import create_engine,Column,String,Integerfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerUSER = "root"PWD = "root"HOST = "172.16.34.34"PORT = "3306"DB = "test"conn_str = 'mysql+pymysql://{}:{}@{}:{}/{}'.format( USER,PWD,HOST,PORT,DB)engine = create_engine(conn_str, echo=True)Base = declarative_base()# 创建实体类class Student(Base): # 指定表名,必须指定 __tablename__ = 'student' # 定义类属性对应字段 id = Column(Integer,autoincrement=True,primary_key=True) name = Column(String(64),nullable=False) age = Column("age",Integer,nullable=False) gen = Column("gender",String(20)) # 第一参数是字段名，如果和属性名不一致，一定要指定 # age = Column('age', Integer) def __repr__(self): return "{} id={} name={} age={} gender={}".format( self.__class__.__name__, self.id, self.name, self.age, self.gen)print(Student)print(repr(Student.__table__))s = Student(name='tom',age=20)print(s.name)s.gender = "M"print(s.gender)# Base.metadata.drop_all(bind=engine)# Base.metadata.create_all(bind=engine) # metadata 记录的是继承自Base的表，删除也是Session = sessionmaker(bind=engine) # 不做操作不会处理连接session =Session() # 创建session对象，也不会在此时连接数据库
CRUD操作
增
add()：增加一个对象
add_all()：可迭代对象，元素是对象session.add(s) #添加一次对象s，未提交print(s)session.commit() #提交一次对象sprint(s) try: session.add_all([s]) print(s) session.commit() #在同一个线程内，再次提交同一个未修改过的对象不会执行except: session.rollback() # 使用要加日志记录 raise
add_all()方法不会提交成功的，不是因为它不对，而是s，s成功提交后，s的主键就有了值，所以，只要s没有修改过，就认为没有改动。如下，s变化了，就可以提交修改了。
s.name = 'jerry' # 修改session.add_all([s])
s主键没有值，就是新增；主键有值，就是找到主键对应的记录修改。
简单查询
使用query()方法，返回一个Query对象students = session.query(Student) # 无条件,相当于select * from studentfor student in students: print(student)print('~~~~~~~~~~~~~')student = session.query(Student).get(3) # 通过主键查询,相当于select * from student where id=3print(student) query方法将实体类传入，返回类的对象可迭代对象，这时候并不查询。迭代它就执行SQL来查询数据库，封装数据到指定类的实例。
get方法使用主键查询，返回一条传入类的一个实例。对象不存在，返回None改
先查后改
student = session.query(Student).get(3)print(student)student.name = 'sam'student.age = 30print(student)session.add(student)session.commit()
删
先查后删
尝试执行以下代码，会发现产生一个异常，未持久的异常
try: student = Student(id=2, name="serry", age=10) session.delete(student) session.commit()except Exception as e: session.rollback() print('~~~~~~~~') print(e) Instance '' is not persisted
数据库查询数据慢且忙，最好在容器中就记录下来，不要重复查询相同的数据
状态
每一个实体，都有一个状态属性_sa_instance_state，其类型是sqlalchemy.orm.state.InstanceState，可以使用sqlalchemy.inspect(entity)函数查看状态。
常见的状态值有transient、pending、persistent、deleted、detached。| 状态 | 说明 |
| transient | 实体类尚未加入到session中，同时并没有保存到数据库中 |
| pending | transient的实体被add()到session中，状态切换到pending，但它还没有flush到数据库中 |
| persistent | session中的实体对象对应着数据库中的真实记录。pending状态在提交成功后可以变成persistent状态，或者查询成功返回的实体也是persistent状态 |
| deleted | 实体被删除且已经flush但未commit完成。事务提交成功了，实体变成detached，事务失败， 返回persistent状态 |
| detached | 删除成功的实体进入这个状态 |
新建一个实体，状态是transient临时的。
一旦add()后从transient变成pending状态。
成功commit()后从pending变成persistent状态。
成功查询返回的实体对象，也是persistent状态。
persistent状态的实体，修改依然是persistent状态。
persistent状态的实体，删除后，flush后但没有commit，就变成deteled状态，成功提交，变为detached状态，提交失败，还原到persistent状态。flush方法，主动把改变应用到数据库中去。
删除、修改操作，需要对应一个真实的记录，所以要求实体对象是persistent状态。
import sqlalchemyfrom sqlalchemy import create_engine,Column,String,Integerfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerUSER = "root"PWD = "root"HOST = "172.16.34.34"PORT = "3306"DB = "test"conn_str = 'mysql+pymysql://{}:{}@{}:{}/{}'.format( USER,PWD,HOST,PORT,DB)engine = create_engine(conn_str, echo=True)Base = declarative_base()# 创建实体类class Student(Base): # 指定表名,必须指定 __tablename__ = 'student' # 定义类属性对应字段 id = Column(Integer,autoincrement=True,primary_key=True) name = Column(String(64),nullable=False) age = Column("age",Integer,nullable=False) gen = Column("gender",String(20)) # 第一参数是字段名，如果和属性名不一致，一定要指定 # age = Column('age', Integer) def __repr__(self): return "{} id={} name={} age={} gender={}".format( self.__class__.__name__, self.id, self.name, self.age, self.gen)print(Student)print(repr(Student.__table__))Session = sessionmaker(bind=engine)session =Session()from sqlalchemy.orm.state import InstanceStatedef getstate(entity,i): insp = sqlalchemy.inspect(entity) state = "session={}, attached={},\ntransient={},persistent={},\npending={},deleted={}.detached={}".format( insp.session_id, insp._attached, insp.transient, insp.persistent, insp.pending, insp.deleted, insp.detached ) print(i,state) print(insp.key) print("_"*30)# student = session.query(Student).get(3)# getstate(student,1)try: student = Student(name="Tony", age=30) getstate(student, 2) # transit student = Student(name="sammy", age=30) getstate(student, 3) # transit session.add(student) # add后变成pending getstate(student, 4) # pending # session.delete(student) # 删除的前提是persistent，否则抛异常 # getstate(student, 5) session.commit() getstate(student, 6) # persistent session.delete(student) # 删除的前提是persistent，否则抛异常 getstate(student, 7) session.flush() getstate(student,8) session.commit() getstate(student, 9)except Exception as e: session.rollback() print('~~~~~~~~') print(e) 返回结果
2 session=None, attached=False,transient=True,persistent=False,pending=False,deleted=False.detached=FalseNone______________________________3 session=None, attached=False,transient=True,persistent=False,pending=False,deleted=False.detached=FalseNone______________________________4 session=1, attached=True,transient=False,persistent=False,pending=True,deleted=False.detached=FalseNone______________________________2018-11-12 18:04:27,682 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode'2018-11-12 18:04:27,682 INFO sqlalchemy.engine.base.Engine {}2018-11-12 18:04:27,683 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'lower_case_table_names'2018-11-12 18:04:27,683 INFO sqlalchemy.engine.base.Engine {}2018-11-12 18:04:27,685 INFO sqlalchemy.engine.base.Engine SELECT DATABASE()2018-11-12 18:04:27,685 INFO sqlalchemy.engine.base.Engine {}2018-11-12 18:04:27,686 INFO sqlalchemy.engine.base.Engine show collation where `Charset` = 'utf8mb4' and `Collation` = 'utf8mb4_bin'2018-11-12 18:04:27,686 INFO sqlalchemy.engine.base.Engine {}2018-11-12 18:04:27,687 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS CHAR(60)) AS anon_12018-11-12 18:04:27,687 INFO sqlalchemy.engine.base.Engine {}2018-11-12 18:04:27,688 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS CHAR(60)) AS anon_12018-11-12 18:04:27,688 INFO sqlalchemy.engine.base.Engine {}2018-11-12 18:04:27,689 INFO sqlalchemy.engine.base.Engine SELECT CAST('test collated returns' AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin AS anon_12018-11-12 18:04:27,689 INFO sqlalchemy.engine.base.Engine {}2018-11-12 18:04:27,690 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)2018-11-12 18:04:27,691 INFO sqlalchemy.engine.base.Engine INSERT INTO student (name, age, gender) VALUES (%(name)s, %(age)s, %(gender)s)2018-11-12 18:04:27,691 INFO sqlalchemy.engine.base.Engine { 'name': 'sammy', 'age': 30, 'gender': None}2018-11-12 18:04:27,692 INFO sqlalchemy.engine.base.Engine COMMIT6 session=1, attached=True,transient=False,persistent=True,pending=False,deleted=False.detached=False( , (11,), None)______________________________7 session=1, attached=True,transient=False,persistent=True,pending=False,deleted=False.detached=False( , (11,), None)______________________________2018-11-12 18:04:27,693 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)2018-11-12 18:04:27,694 INFO sqlalchemy.engine.base.Engine SELECT student.gender AS student_gender, student.id AS student_id, student.name AS student_name, student.age AS student_age FROM student WHERE student.id = %(param_1)s2018-11-12 18:04:27,694 INFO sqlalchemy.engine.base.Engine { 'param_1': 11}2018-11-12 18:04:27,695 INFO sqlalchemy.engine.base.Engine DELETE FROM student WHERE student.id = %(id)s2018-11-12 18:04:27,695 INFO sqlalchemy.engine.base.Engine { 'id': 11}8 session=1, attached=True,transient=False,persistent=False,pending=False,deleted=True.detached=False( , (11,), None)______________________________2018-11-12 18:04:27,696 INFO sqlalchemy.engine.base.Engine COMMIT9 session=None, attached=False,transient=False,persistent=False,pending=False,deleted=False.detached=True( , (11,), None)______________________________
复杂查询
实体类
import enumclass GenderEnum(enum.Enum): M = "M" F = "F"class Employee(Base): __tablename__ = "employees" emp_no = Column(Integer, primary_key=True) birth_date = Column(Date, nullable=False) first_name = Column(String(14), nullable=False) last_name = Column(String(16), nullable=False) gender = Column(Enum(GenderEnum), nullable=False) hire_date = Column(Date, nullable=False) def __repr__(self): return "{} no={} name={} {} gender={}".format( self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender.value ) # 返回的迭代器，查看内容def show(emps): for x in emps: print(x) 以下语句为条件
最简单的查询：
emps = session.query(Employee).filter(Employee.emp_no > 10015)show(emps)
与，或，非
and条件可以使用两个filter实现，也可以使用and_，也可使用运算符重载 &
emps = session.query(Employee).filter(Employee.emp_no > 10015).filter(Employee.gender ==GenderEnum.F)show(emps)emps = session.query(Employee).filter(and_(Employee.emp_no > 10015, Employee.gender ==GenderEnum.M))show(emps)#运算符重载注意表达式要加括号emps = session.query(Employee).filter((Employee.emp_no > 10015) & (Employee.gender == GenderEnum.M))show(emps)
or 条件可以使用or_ 或者运算符 |
emps = session.query(Employee).filter(or_(Employee.emp_no > 10018, Employee.emp_no < 10003))show(emps)#加括号emps = session.query(Employee).filter((Employee.emp_no > 10018) | (Employee.emp_no < 10003))show(emps)
not 条件使用not_ 或者运算符 ~
emps = session.query(Employee).filter(not_(Employee.emp_no < 10018))show(emps)#加括号emps = session.query(Employee).filter(~(Employee.emp_no < 10018))show(emps)
总之，与或非的运算符&、|、~，一定要在表达式上加上括号
in
emplist = [10010, 10015, 10018]emps = session.query(Employee).filter(Employee.emp_no.in_(emplist))show(emps)
not in
emplist = [10010, 10015, 10018]emps = session.query(Employee).filter(~Employee.emp_no.in_(emplist))show(emps)emps = session.query(Employee).filter(Employee.emp_no.notin_(emplist))show(emps)
like ,少用
emps = session.query(Employee).filter(Employee.last_name.like('P%'))show(emps) not like ,少用
emps = session.query(Employee).filter(Employee.last_name.notlike('P%')) ilike 忽略大小写
emps = session.query(Employee).filter(Employee.last_name.ilike('P%'))show(emps)
排序
升序
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no)emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.asc())show(emps)
降序
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.desc())show(emps)
多列排序
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.last_name).order_by(Employee.emp_no.desc())show(emps)emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.last_name.desc(), Employee.emp_no.asc())show(emps)
分页
emps = session.query(Employee).limit(4)show(emps) emps = session.query(Employee).limit(4).offset(18)show(emps)
消费者方法
总共的行数
emps = session.query(Employee)print(len(list(emps))) # 返回大量的结果集，然后转换listprint(emps.count()) # 聚合函数count(*)的查询
取所有数据
emps = session.query(Employee)print(emps.all()) # 返回列表，查不到返回空列表
取首行 ： first方法本质上就是limit语句
emps = session.query(Employee)print(emps.limit(1).one()) #返回一行print(emps.one()) #如果查询结果是多行抛异常
删除
# 删除 delete by querysession.query(Employee).filter(Employee.emp_no > 10018).delete()#session.commit() # 提交则删除
聚合，分组
聚合
from sqlalchemy import funcquery = 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(Employee.gender,func.count(Employee.emp_no)).group_by(Employee.gender).all())
关联查询
有两张表，其中的对象多对多关联，就要建立第三张表。
如果是一对多关联，就在多的一端建立外键。
有一个员工，即属于A部门，又属于B部门，同时每个部门都有许多员工，这就是多对多
先把这些表的Model类和字段属性建立起来。
class Employee(Base):# 指定表名 __tablename__ = 'employees' # 定义属性对应字段 emp_no = Column(Integer, primary_key=True) birth_date = Column(Date, nullable=False) first_name = Column(String(14), nullable=False) last_name = Column(String(16), nullable=False) gender = Column(Enum(GenderEnum), nullable=False) hire_date = Column(Date, nullable=False) # 第一参数是字段名，如果和属性名不一致，一定要指定 # age = Column('age', Integer) def __repr__(self): return "{} no={} name={} {} gender={}".format( self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender.value )class Department(Base): __tablename__ = 'departments' dept_no = Column(String(4), primary_key=True) dept_name = Column(String(40), nullable=False, unique=True) def __repr__(self): return "{} no={} name={}".format(type(self).__name__, self.dept_no, self.dept_name)class Dept_emp(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 "{} empno={} deptno={}".format(type(self).__name__, self.emp_no, self.dept_no)
查询10010员工的所在部门和标号信息
results = session.query(Employee,Dept_emp).filter((Employee.emp_no == Dept_emp.emp_no) & (Employee.emp_no == 10010)).all()show(results)
这种方式产生隐式连接的语句
使用join
results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == 10010).all() results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010).all()print(results)
返回的都只有一行数据。它们生成的SQL语句是一样的，执行该SQL语句返回确实是2行记录，可以Python中的返回值列表中只有一个元素。
原因在于 query(Employee) 这个只能返回一个实体对象中去，为了解决这个问题，需要修改实体类Employee，增加属性用来存放部门信息
sqlalchemy.orm.relationship(实体类名字符串)
from sqlachemy import relationshipclass Employee(Base):# 指定表名 __tablename__ = 'employees' # 定义属性对应字段 emp_no = Column(Integer, primary_key=True) birth_date = Column(Date, nullable=False) first_name = Column(String(14), nullable=False) last_name = Column(String(16), nullable=False) gender = Column(Enum(GenderEnum), nullable=False) hire_date = Column(Date, nullable=False) # 第一参数是字段名，如果和属性名不一致，一定要指定 # age = Column('age', Integer) departments = relationship("Dept_emp") def __repr__(self): return "{} no={} name={} {} gender={} depts={}".format( self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender.value,self.departments ) 查询
# 第一种# results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010)# 第二种# results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010)# 第三种results = session.query(Employee).join(Dept_emp, (Employee.emp_no == Dept_emp.emp_no) & (Employee.emp_no == 10010))show(results.all())
第一种方法join(Dept_emp)中没有等值条件，会自动生成一个等值条件，如果后面有filter，哪怕是filter(Employee.emp_no == Dept_emp.emp_no)，这个条件会在where中出现。第一种这种自动增加join的等值条件的方式不好，不要这么写
第二种方法在join中增加等值条件，阻止了自动的等值条件的生成。这种方式推荐第三种方法就是第二种，这种方式也可以只要不访问departments属性，就不会查dept_emp这张表。
总结
在开发中，一般都会采用ORM框架，这样就可以使用对象操作表了。定义表映射的类，使用Column的描述器定义类属性，使用ForeignKey来定义外键约束。
如果在一个对象中，想查看其它表对应的对象的内容，就要使用relationship来定义关系。
是否使用外键约束？
1、力挺派 能使数据保证完整性一致性2、嫌弃派 开发难度增加，大数据的时候影响插入、修改、删除的效率。 在业务层保证数据的一致性。