介绍

SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果，并把获取的结果转为python对象。其中发sql到mysql服务器，从mysql服务器拿结果都是借助其他工具来完成的，例如pymysql.

Engine，框架的引擎
Connection Pooling ，数据库连接池
Dialect，选择连接数据库的DB API种类
Schema/Types，架构和类型
SQL Exprression Language，SQL表达式语言

SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

1MySQL-Python
2    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
3    
4pymysql
5    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
6    
7MySQL-Connector
8    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
9    
10cx_Oracle
11    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
12    
13更多：http://docs.sqlalchemy.org/en/latest/dialects/index.html

单表

单表的创建

1import datetime
2import time
3
4from sqlalchemy.ext.declarative import declarative_base
5from sqlalchemy import create_engine
6from sqlalchemy import Column
7from sqlalchemy import Integer, String, Date
8
9from sqlalchemy.orm import sessionmaker
10
11
12Base = declarative_base()
13
14engine = create_engine(
15    "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8",
16    encoding='utf8',
17    max_overflow=0,
18    pool_size=5,
19    pool_timeout=20,
20    pool_recycle=-1
21)
22
23class User(Base):
24    # __tablename__ 字段必须有，否则会报错
25    __tablename__ = 'user'
26    # 不同于django model 会自动加主键，sqlalchemy需要手动加主键
27    id = Column(Integer, primary_key=True)
28    name = Column(String(32), nullable=False)
29    # 时间类型的default的默认值需要使用datetime.date.today(), 但是使用flask-sqlalchemy的时候使用datetime.date.today
30    date = Column(Date, default=datetime.date.today())
31
32def create_table():
33    # 创建所有的表,表如果存在也不会重复创建，只会创建新的表，而且sqlalchemy默认不支持修改表结构
34    # 要想和django orm一样能修改表结构并反映到数据库需要借助第三方组件
35    Base.metadata.create_all(engine)
36
37def drop_table():
38    # 删除所有的表
39    Base.metadata.drop_all(engine)

单表的增删改查

1# 增加
2# user = User(name='jack')
3# session.add(user)
4# session.commit()
5# session.close()
6# # 增加多条
7# user_list = [User(name='a'), User(name='b'), User(name='c')]
8# session.add_all(user_list)
9# session.commit()
10
11# 查
12
13# result 是一个列表，里面存放着对象
14# result = session.query(User).all()
15# for item in result:
16#     print(item.name)
17
18# 查询最后加all() 得到的是一个存放对象的列表，不加all() 通过print 打印出的是sql语句
19# 但是结果仍是一个可迭代的对象，只不过对象的__str__ 返回的是sql语句，迭代的时候里面的对象
20# 是一个类元组的对象，可以使用下标取值，也可以通过对象的`.`方式取值
21# result = session.query(User.name, User.date).filter(User.id>3)
22# for item in result:
23#     print(item[0], item.date)
24
25# 条件查询
26from sqlalchemy import and_, or_,func
27
28## 逻辑查询
29r0 = session.query(User).filter(User.id.in_([1, 2]))
30r1 = session.query(User).filter(~User.id.in_([1, 2]))
31r2 = session.query(User).filter(User.name.startswith('j'), User.id>2)
32r3 = session.query(User).filter(
33    or_(
34        User.id>3,
35        and_(User.name=='jack', User.id<2)
36    )
37)
38
39## 通配符
40r4 = session.query(User).filter(User.name.like('%j'))
41r5 = session.query(User).filter(~User.name.like('%j'))
42
43## limit 和django orm 一样都是通过索引来限制
44r6 = session.query(User)[0:4]
45
46## 排序, 排序一般是倒数第二的位置，倒数第一是limit
47r7 = session.query(User).order_by(User.id.desc())
48
49## 分组和聚合
50r8 = session.query(func.max(User.id)).group_by(User.name).all()
51
52# 改, 得到的结果是收到影响的记录条数
53# r9 = session.query(User).filter(User.id==2).update({'name': User.name + User.name.concat('hh')}, synchronize_session=False)
54
55# 删除
56session.query(User).delete()
57
58## 子查询
59session.query(User).filter(User.id.in_(session.query(User.id).filter(User.name.startswith('j'))))
60
61session.commit()
62# 这边的close并不是真实的关闭连接，而是完成终止事务和清除工作
63session.close()

连表

两张表

创建表

1import datetime
2import time
3
4from sqlalchemy.ext.declarative import declarative_base
5from sqlalchemy import create_engine
6from sqlalchemy import Column
7from sqlalchemy import Integer, String, Date
8from sqlalchemy import ForeignKey
9from sqlalchemy.orm import sessionmaker
10
11
12Base = declarative_base()
13
14engine = create_engine(
15    "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8",
16    encoding='utf8',
17    max_overflow=0,
18    pool_size=5,
19    pool_timeout=20,
20    pool_recycle=-1
21)
22
23class User(Base):
24    # __tablename__ 字段必须有，否则会报错
25    __tablename__ = 'user'
26    # 不同于django model 会自动加主键，sqlalchemy需要手动加主键
27    id = Column(Integer, primary_key=True)
28    name = Column(String(32), nullable=False)
29    # 时间类型的default的默认值需要使用datetime.date.today(), 但是使用flask-sqlalchemy的时候使用datetime.date.today
30    date = Column(Date, default=datetime.date.today())
31
32    # 因为外键的sh设置更偏向于数据库底层，所以这里使用了表名，而不是类名
33    depart_id = Column(Integer, ForeignKey('department.id'))
34
35class Department(Base):
36    __tablename__ = 'department'
37    id = Column(Integer, primary_key=True)
38    # 默认的nullable 是True
39    title = Column(String(32), nullable=False)

查询

1# 默认根据在类里面定义的外键进行on， 此时得到的结果是[(userobj, departmnetobj),()] 这种形式，默认是inner join
2r1 = session.query(User, Department).join(Department).all()
3r2 = session.query(User.name, Department.title).join(Department, Department.id==User.depart_id).all()
4
5# 有了 isouter 参数，inner join 就变成 left join
6r3 = session.query(User.name, Department.title).join(Department, Department.id==User.depart_id, isouter=True).all()

relationship

现在问题来了，想要查name是jack所属的部门名，两种方式

分两次sql查询

user = session.query(User).filter(User.name == 'jack').first() title = session.query(Department.title).filter(Department.id == user.depart_id).first().title
一次连表查询

r1 = session.query(Department.title).join(User).filter(User.name == 'jack').first().title print(r1)

这样的方式在python代码的级别貌似没有django的方便，django 的 orm 拿到一个对象obj， obj.deaprtment.title 就能拿到结果。sqlalchemy也有类似功能，通过relationship来实现。

1# 注意，导入的是relationship,而不是relationships
2from sqlalchemy.orm import relationship
3class Department(Base):
4    __tablename__ = 'department'
5    id = Column(Integer, primary_key=True)
6    # 默认的nullable 是True
7    title = Column(String(32), nullable=False)
8
9    # 如果backref 的那张表和这张表是一对一关系，加上一个uselist=False参数就行
10    user = relationship("User", backref='department')
11
12class User(Base):
13    # __tablename__ 字段必须有，否则会报错
14    __tablename__ = 'user'
15    # 不同于django model 会自动加主键，sqlalchemy需要手动加主键
16    id = Column(Integer, primary_key=True)
17    name = Column(String(32), nullable=False)
18    # 时间类型的default的默认值需要使用datetime.date.today(), 但是使用flask-sqlalchemy的时候使用datetime.date.today
19    date = Column(Date, default=datetime.date.today())
20
21    # 因为外键的sh设置更偏向于数据库底层，所以这里使用了表名，而不是类名
22    depart_id = Column(Integer, ForeignKey('department.id'))
23    # 神奇的一点是，SQLAlchemy会根据关系的对应情况自动给关系相关属性的类型
24    # 比如这里的Department下面的user自动是一个list类型，而User由于设定了外键的缘故
25    # 一个user最多只能应对一个用户，所以自动识别成一个非列表类型
26    # 这样写两个relationship比较麻烦，在设置了外键的一边使用relationship，并且加上backref参数
27    # department = relationship("Department")
28
29
30
31session_factory = sessionmaker(engine)
32session = session_factory()
33
34user = session.query(User).first()
35print(user.department)
36
37department = session.query(Department).first()
38print(department.user)

有了relationship，不仅查询方便，增加数据也更方便。

1# 增加一个用户ppp，并新建这个用户的部门叫IT
2
3## 方式一
4# d = Department(title='IT')
5# session.add(d)
6# session.commit()  # 只有commit之后才能取d的id
7#
8# session.add(User(name='ppp', depart_id=d.id))
9# session.commit()
10
11## 方式二
12
13# session.add(User(name='ppp', department=Department(title='IT')))
14# session.commit()
15
16# 增加一个部门xx，并在部门里添加员工：aa/bb/cc
17# session.add(Department(title='xx', users=[User(name='aa'), User(name='bb'),User(name='cc')]))
18# session.commit()

三张表

创建表

1from sqlalchemy.orm import relationship
2from sqlalchemy.ext.declarative import declarative_base
3from sqlalchemy import create_engine
4from sqlalchemy import Column
5from sqlalchemy import Integer, String, Date
6from sqlalchemy import ForeignKey, UniqueConstraint, Index
7class Student(Base):
8    __tablename__ = 'student'
9    id = Column(Integer, primary_key=True)
10    name = Column(String(32), index=True, nullable=False)
11
12    course_list = relationship('Course', secondary='student2course', backref='student_list')
13
14class Course(Base):
15    __tablename__ = 'course'
16    id = Column(Integer, primary_key=True)
17    title = Column(String(32), index=True, nullable=False)
18
19class Student2Course(Base):
20    __tablename__ = 'student2course'
21    id = Column(Integer, primary_key=True, autoincrement=True)
22    student_id = Column(Integer, ForeignKey('student.id'))
23    course_id = Column(Integer, ForeignKey('course.id'))
24
25    __table_args__ = (
26        UniqueConstraint('student_id', 'course_id', name='uix_stu_cou'), # 联合唯一索引
27        # Index('student_id', 'course_id', name='stu_cou'),                          # 联合索引
28    )

查询

查询方式和只有两张表的情况类似，例如查询jack选择的所有课

1# obj = session.query(Student).filter(Student.name=='jack').first()
2# for item in obj.course_list:
3#     print(item.title)

创建一个课程，创建2学生，两个学生选新创建的课程

1# obj = Course(title='英语')
2# obj.student_list = [Student(name='haha'),Student(name='hehe')]
3#
4# session.add(obj)
5# session.commit()

执行原生sql

方式一

1# 查询
2# cursor = session.execute('select * from users')
3# 拿到的结果是一个ResultProxy对象，ResultProxy对象里套着类元组的对象，这些对象可以通过下标取值，也可以通过对象.属性的方式取值
4# result = cursor.fetchall()
5
6# 添加
7cursor = session.execute('INSERT( INTO users(name) VALUES(:value)', params={"value": 'wupeiqi'})
8session.commit()
9print(cursor.lastrowid)

方式二

1import pymysql
2conn = engine.raw_connection()
3cursor = conn.cursor(pymysql.cursors.DictCursor)
4cursor.execute(
5    "select * from user"
6)
7result = cursor.fetchall()
8# 结果是一个列表，列表里面套着的对象就是原生的字典对象
9print(result)
10cursor.close()
11conn.close()

多线程情况下的sqlalchemy

在每个线程内部创建session并关闭session

1session_factory = sessionmaker(engine)
2
3def task(i):
4    # 创建一个会话对象，没错仅仅是创建一个对象这么简单
5    session = session_factory()
6    # 执行query语句的时候才会真真去拿连接去执行sql语句，如果没有close那么没有空闲连接就会等待
7    result = session.execute('select * from user where id=14')
8    for i in result:
9        print(i.name)
10    time.sleep(1)
11    # 必须要close,这里的close可以理解为关闭会话，把链接放回连接池
12    # 如果注释掉这一句代码，程序会报错QueuePool limit of size 5 overflow 0 reached, connection timed out, timeout 20
13    session.close()
14
15if __name__ == '__main__':
16    for i in range(10):
17        t = Thread(target=task, args=(i,))
18        t.start()

结果是每5个一起打印

在全局创建一个特殊的session，各个线程去使用这个特殊的session

1from sqlalchemy.orm import scoped_session
2
3session_factory = sessionmaker(engine)
4session = scoped_session(session_factory)
5
6def task(i):
7    result = session.execute('select * from user where id=14')
8    for i in result:
9        print(i.name)
10    time.sleep(1)
11
12    session.remove()
13
14if __name__ == '__main__':
15    for i in range(10):
16        t = Thread(target=task, args=(i,))
17        t.start()

scoped_session 这个类还真是神奇，名字竟然还不是大写，而且原先的session有的，这个类实例化的对象也会有。我们第一反应是继承，其实它也不是继承。它的实现原理是这样的执行导入语句的from sqlalchemy.orm import scoped_session的时候，点进去看源码发现执行了一个scoping.py的文件。

最终self.registry()就是session_factory() 对象，而且是线程隔离的，每个线程有自己的会话对象

SQLAlchemy

介绍

单表