ORM_declarative_base

ORM，对象关系映射，对象和关系之间的映射，使用面向对象的方式来操作数据库

关系模型和Python对象之间的映射
table => class         表映射为类
row =>object          行映射为实例
column=> property  字段映射为属性
1
2
3
4

SQLAIchemy

SQLAIchemy是一个ORM框架

• 安装

pip install sqlalchemy
1

开发

• 创建连接
  SQLAIchemy内部使用连接池
  数据库连接的事情，交给引擎

dialect+driver:// username:password@host:port/database
# mysqldb连接
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
engine=sqlalchemy.create_engine("mysql+mysqldb://user:password@host:port/database")
#pymysql的连接
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
engine=sqlalchemy.create_engine("mysql+mysqldb://user:password@host:port/database")
engine=sqlalchemy.create_engine("mysql+mysqldb://user:password@host:port/database",echo=True)
echo=True
引擎是否打印执行的语句，调试的时候打开很方便
lazy connecting:懒连接，创建引擎并不会马上连接数据库，直到让数据库执行任务时才连接
1
2
3
4
5
6
7
8
9
10
11

• Declace a Mapping创建映射

1、创建基类

from sqlalchemy.ext.declarative import declarative_base
Base=declarative_base()
1
2

2、创建实体类
student表

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String
engine=sqlalchemy.create_engine("mysql+pymysql://xiaobai:xiaobai@192.168.163.130:3306/school",echo=True)
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)
    def __repr__(self):
        return "{} id={} name={} age={}".format(self.__class__.__name__,self.id,self.name,self.age)
print(Student)
print(repr(Student.__table__))
1
2
3
4
5
6
7
8
9
10
11
12
13
14

__tablename__指定表名
Column类指定对应的字段，必须指定
3、创建表
可以使用SQLAlchemy来创建、删除表

#删除继承自Base的所有表
Base.metadata.drop_all(engine)
#创建继承自Base的所有表
Base.metadata.create_all(engine)
1
2
3
4

生产环境很少这样创建表，都是系统上线的时候由脚本生成
生成环境很少删除表，宁可废弃都不能删除

4、创建会话session
在一个会话中操作数据库，会话建立在连接上，连接被引擎管理。
当第一次使用数据库时，从引擎维护的连接池中获取一个连接使用

from sqlalchemy.orm import sessionmaker
#创建session
Session=sessionmaker(bind=engine) #工厂方法返回类
session=Session()  #实例化
#依然在第一次使用时连接数据库
1
2
3
4
5

session对象线程不安全，所以不同线程应该使用不同的session对象
Session类和engine有一个就可以了

CRUD操作

增
add():增加一个对象
add_all():可迭代对象，元素是对象

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String
from sqlalchemy.orm import sessionmaker
engine=sqlalchemy.create_engine("mysql+pymysql://xiaobai:xiaobai@192.168.163.132:3306/test",echo=True)
# 创建基类
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)
    def __repr__(self):
        return "<{}{}{}>".format(self.__class__.__name__,self.id,self.name,self.age)
#删除表
Base.metadata.drop_all(engine)
#创建表
Base.metadata.create_all(engine)

#创建session
Session=sessionmaker(bind=engine)
session=Session()

s=Student(name='tom') # 构造时传入
s.age=20   #属性赋值
print(s)

session.add(s)
print(s)
session.commit()
print('`````````````````````````````')
try:
    session.add_all([s])
    print(s,'````````````````````````')
    session.commit()
    print(s)
except:
    session.collback()
    print('roll back')
    raise
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

add_all()方法不会提交成功的，不是因为它不对，而是s，s成功提交后，s的主键就有了值，所以只要s没有修改过，就认为没有改动。
s主键没有值，就是新增，主键有值，就是找到主键对应的记录修改。

简单查询
使用query()方法，返回一个Query对象

students=session.query(Student) #无条件
print(students)
for student in students:
    print(repr(student))
print('```````````````````````````')

student=session.query(Student).get(1)
print(student)
1
2
3
4
5
6
7
8

query方法将实体类传入，返回类的对象可迭代对象，这时候并不查询，迭代它就执行SQL来查询数据库，封装数据到指定类的实例
get方法使用主键查询，返回一条传入类的一个实例
改

student=session.query(Student).get(2)
print(student)
student.name='ergou'
student.age=50
print(student)
session.add(student)
session.commit()
1
2
3
4
5
6
7

先查回来，修改后，再提交更改

删除

try:
    student=Student(id=2,name="sam",age=30)
    session.delete(student)
    session.commit()
except Exception as e:
    session.rollback()
    print('``````````````')
    print(e)
1
2
3
4
5
6
7
8

会产生一个异常

Instance '<Student at 0x1f997aa9748>' is not persisted 未持久化异常！
1

状态
每一个实体，都有一个状态属性_sa_instance_state，其类型是sqlalchemy.orm.state.InstanceState，可以使用sqlalchemy.inspect(entity)函数查看状态
常见的状态值有transient(短暂)、pending(未决定)、persistent(持久)、deleted(已删除)、detached(分离)

新建一个实体，状态是transient临时的
一旦add()后从transient变成pending状态
成功commit()后从pending变成persistent状态
成功查询返回的实体对象，也是persistent状态
persistent状态的实体，修改依然是persistent状态
persistent状态的实体，删除后，flush后但没有commit，就变成detele状态，成功提交，变为detached状态，提交失败，还原到persistent状态，flush方法，主动把概念应用大数据库中去。
删除、修改操作，需要对应一个真实的记录，所以要求实体对象是persistent状态

import sqlalchemy
from sqlalchemy import create_engine,Column,Integer,String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

connsrt="{}://{}:{}@{}:{}/{}".format('mysql+pymysql','xiaobai','xiaobai','192.168.163.133',3306,'test')
# 创建连接
engine=create_engine(connsrt,echo=True)

# 创建基类
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)

    def __repr__(self):
        return "{} id={} name={} age={}".format(__class__.__name__,self.id,self.name,self.age)
# 创建会话
Session=sessionmaker(bind=engine)
session=Session()
from sqlalchemy.orm.state import InstanceState
def getstate(instance,i):
    inp: InstanceState=sqlalchemy.inspect(instance)
    states="{}:key={},nsid={},attached={},transient={},pending={}\n,persistent={},deleted={},detached={}".format(
        i,inp.key,
        inp.session_id,inp._attached,inp.transient,
        inp.pending,inp.persistent,inp.deleted,inp.detached
    )
    print(states,end='\n------------------------\n')
student=session.query(Student).get(2)
getstate(student,1)

try:
    student=Student(id=2,name='ergou',age=50)
    getstate(student,2)

    student=Student(name='sammy',age=30)
    getstate(student,3)
    session.add(student)
    getstate(student,4)

    session.commit()
    getstate(student,5)

    session.delete(student)
    getstate(student,6)
    session.commit()
    getstate(student,7)
except Exception as e:
    session.rollback()
    print(e,'````````````````')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

复杂查询

from sqlalchemy import Column,Integer,String,Date,Enum,ForeignKey,create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import enum
Base=declarative_base()
connstr="{}://{}:{}@{}:{}/{}".format(
    "mysql+pymysql",'xiaobai','xiaobai',
    "192.168.163.133",3306,'test'
)
engine=create_engine(connstr,echo=True)
Session=sessionmaker(bind=engine)
session=Session()

class MyEnum(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(MyEnum),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)
    print("````````````````````")

#简单查询
emps=session.query(Employee).filter(Employee.emp_no>10015)
show(emps)
# 与或非
from sqlalchemy import or_,and_,not_
# AND条件
emps=session.query(Employee).filter(Employee.emp_no>10015).filter(Employee.gender==MyEnum.F)
show(emps)

emps=session.query(Employee).filter(Employee.emp_no>10015,Employee.emp_no<10018)
show(emps)

emps=session.query(Employee).filter(and_(Employee.emp_no>10015,Employee.gender==MyEnum.M))
show(emps)

emps=session.query(Employee).filter((Employee.emp_no>10015)&(Employee.gender==MyEnum.F))
show(emps)
# OR条件
emps=session.query(Employee).filter((Employee.emp_no>10018)|(Employee.emp_no<10003))
show(emps)

emps=session.query(Employee).filter(or_(Employee.emp_no>10018,Employee.emp_no<10003))
show(emps)

# 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
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%'))
show(emps)

# 排序
emps=session.query(Employee).filter(Employee.emp_no>10010).order_by(Employee.emp_no)
show(emps)

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).limit(4)
show(emps)

emps=session.query(Employee).limit(4).offset(18)
show(emps)

# 总行数
emps=session.query(Employee)
print(len(list(emps)))
print(emps.count())

# 取所有数据
print(emps.all())

# 取行首
print(emps.first())

#有且只能有一行
print(emps.limit(1).one())

#删除
session.query(Employee).filter(Employee.emp_no>10018).delete()

# 聚合函数
#count
from sqlalchemy import func

query=session.query(func.count(Employee.emp_no))
# 列表中一个元素
print('``````````````',query.all(),'````````````````````')
#一个只有一个元素的元组
print(query.first())
# 只能有一行返回，一个元组
print(query.one())
# 取one()的第一个元素
print(query.scalar())

# 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())

# 分组
query=session.query(Employee.gender,
                    func.count(Employee.emp_no)).group_by(Employee.gender).all()
for g,y in query:
    print(g,g.value,y)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

关联查询

from sqlalchemy import Column,create_engine,Integer,String,Enum,Date,ForeignKey
from  sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship
import sqlalchemy,enum
Base=declarative_base()
# 创建会话
engine=create_engine("mysql+pymysql://xiaobai:xiaobai@192.168.163.134:3306/test",echo=True)
Session=sessionmaker(bind=engine)
session=Session()
class Myenum(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(Myenum), nullable=False)
     hire_date = Column(Date, nullable=False)
     departments = relationship('Dept_emp')
     # 第一参数是字段名，如果和属性名不一致，一定要指定
     # age = Column('age', Integer)
     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)
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)

def show(emps):
    for x in emps:
        print(x)
    print("````````````````````")
results=session.query(Employee,Dept_emp).filter(Employee.emp_no==Dept_emp.emp_no).filter(
    Employee.emp_no==10010).all()
show(results)

results=session.query(Employee).join(Dept_emp).filter(Employee.emp_no==10010).all()
show(results)

results=session.query(Employee).join(Dept_emp,Employee.emp_no==Dept_emp.emp_no).filter(Employee.emp_no
                                                                                  ==10010).all()
print(results)

# 查询10010员工的所在的部门编号及其员工信息
results=session.query(Employee).join(Dept_emp).filter(Employee.emp_no==Dept_emp.emp_no).filter(Employee.emp_no==10010)
show(results)
print(11111111111)

results=session.query(Employee).join(Dept_emp,Employee.emp_no==Dept_emp.emp_no).filter(Employee.emp_no==10010)
show(results)

results=session.query(Employee).join(Dept_emp,(Employee.emp_no==Dept_emp.emp_no)&
                                     (Employee.emp_no==10010))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

第一种方法join(Dept_emp)中没有等值条件，会自动生成一个等值条件，如果后面有filter，哪怕是filter(Employee.emp_no==Ddept_emp.emp_no),这个条件会在where中出现，第一种这种自动增加join的等值条件的方式不好。
第二种在join中增加等值条件，阻止了自动的等值条件的生成

总结

在开发中，一般都会采用ORM框架，这样就可以使用对象操作表了。
定义表映射的类，使用Column的描述器定义类属性，使用ForeignKey来定义外键约束。
如果在一个对象中，想查看其它表对应的对象的内容，就要使用relationship来定义关系。
是否使用外键约束？
1、力挺派
能使数据保证完整性一致性
2、弃用派
开发难度增加，大量数据的时候影响插入、修改、删除的效率。
在业务层保证数据的一致性。