SQLAlchemy 从入门到入门

1. SQLAlchemy 介绍

SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

安装：pip3 install sqlalchemy

组成部分：

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

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

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

django中如何反向生成models

python manage.py inspectdb > app/models.py
1

2. 简单使用（能创建表，删除表，不能修改表）

2.1 使用原生 sql

import threading
from sqlalchemy import create_engine

engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/d8?charset=utf8mb4",
    max_overflow=0,  	# 超过连接池大小外最多创建的连接
    pool_size=5,  		# 连接池大小
    pool_timeout=30,  	# 池中没有线程最多等待的时间，否则报错
    pool_recycle=-1  	# 多久之后对线程池中的线程进行一次连接的回收（重置）
)


def task(arg):
	# 从连接池中拿一个链接
    conn = engine.raw_connection()
    cursor = conn.cursor()
    cursor.execute(
        "select * from t8"
    )
    result = cursor.fetchall()
    print(result)
    cursor.close()
    conn.close()


for i in range(20):
    t = threading.Thread(target=task, args=(i,))
    t.start()
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

2.2 orm 使用（线程安全）

• 模型表

• orm 不能创建数据库，需要手动创建

• sqlalchemy 只能创建表和删除表，不能新增，删除字段（使用第三方）

import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

# 实例化得到对象
Base = declarative_base()


class Users(Base):
    id = Column(Integer, primary_key=True)  				# id 主键
    name = Column(String(32), index=True, nullable=False)  	# name列，索引，不可为空
    email = Column(String(32), unique=True)  				# 唯一
    # datetime.datetime.now 不能加括号，加了括号，以后永远是当前时间
    ctime = Column(DateTime, default=datetime.datetime.now)
    extra = Column(Text, nullable=True)

    __tablename__ = 'users'  # 设置表名称，不指定则使用函数的名字

    __table_args__ = (		 # 类似于 Django 的 class META
        UniqueConstraint('id', 'name', name='uix_id_name'),  # 联合唯一
        Index('ix_id_name', 'name', 'email'),  				 # 索引
    )


def init_db():
    """
    根据类创建数据库表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:xwx668428@127.0.0.1:3306/flask_sql?charset=utf8mb4",
        max_overflow=0,  	# 超过连接池大小外最多创建的连接
        pool_size=5,  		# 连接池大小
        pool_timeout=30,  	# 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  	# 多久之后对线程池中的线程进行一次连接的回收（重置）
    )

    Base.metadata.create_all(engine)


def drop_db():
    """
    根据类删除数据库表
    """
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/flask_sql?charset=utf8mb4",
        max_overflow=0,  	# 超过连接池大小外最多创建的连接
        pool_size=5,  		# 连接池大小
        pool_timeout=30,  	# 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  	# 多久之后对线程池中的线程进行一次连接的回收（重置）
    )

    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    # 创建所有表
    init_db()
    # 删除所有表
    # drop_db()
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

• app.py（线程安全）

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
from sqlalchemy.orm import scoped_session
from threading import Thread

# 第一步：创建 engine
engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/flask_sql?charset=utf8mb4",
    max_overflow=0,  	# 超过连接池大小外最多创建的连接
    pool_size=5,  		# 连接池大小
    pool_timeout=30,  	# 池中没有线程最多等待的时间，否则报错
    pool_recycle=-1  	# 多久之后对线程池中的线程进行一次连接的回收（重置）
)

# 第二步：生成session,每次执行数据库操作时，都需要创建一个Session。原本是 session = Session()
Session = sessionmaker(bind=engine)

# 使用sessionmaker，就是线程安全，session对象在flask框架中就可以是全局的，任意线程使用的是就是这个全局session，但是不会出现数据错乱问题
# 本质内部使用了local对象，保证每个线程实际上使用的是当前线程自己的session
session = scoped_session(Session)


# scoped_session类的对象，正常来讲是没有 add， close，commit等方法和属性的，实际上是通过create_proxy_methods装饰器，设置进去的（通过反射setattr写进去的）


def task(i):
    # 第三步：插入数据
    user = Users(name='xwx_%s' % i, email='123%s@qq.com' % i, extra='hhh')
    session.add(user)
    # 第四步：提交
    session.commit()
    # 第五步：关闭链接
    session.close()


for i in range(1):
    t = Thread(target=task, args=[i, ])
    t.start()
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

3. 基本增删改查

...
Session = sessionmaker(bind=engine)
session = Session()
1
2
3

3.1 增加数据

1. 增加数据一条
session.add(对象)	# 对象类似于 Users(name='xwx', age='12'), Users 为表名

2. 增加多条数据
session.add_all([对象A, 对象B])
1
2
3
4
5

3.2 删除数据

res = session.query(Users).filter(Users.id == 1).delete()
print(res)	# 打印的是影响的数量
1
2

3.3 修改数据

1. 查询后修改
res = session.query(Users).filter(Users.id > 0).update({"name": "xwx"})
print(res) # 打印的是影响的数量

2. 类似于django的F查询，如果是字符类型设置synchronize_session为False追加字符串。数字类型设置为evaluate进行数值操作
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "xwx"}, synchronize_session=False)
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")
1
2
3
4
5
6
7

3.4 查询

1. 查询所有
res = session.query(Users).all()
print(res)	# 查询的结果

2. 查询指定的字段类似于原生sql语句:  'select name as xx,age from users;'
res = session.query(Users.name.label('xx'), Users.age).all()
print(res)	# 查询的结果
1
2
3
4
5
6
7

3.5 过滤

1. filter 条件过滤，三种基本条件( > < == ) 且支持多条件连用
res = session.query(Users).filter(Users.id > 0, Users.name == 'xwx').all()
res1 = session.query(Users).filter(Users.name == 'xwx').first()

2. filter_by 条件过滤，使用表达式的形式（没有大于小于，且直接使用字段名过滤即可）
res = session.query(Users).filter_by(name='xwx', email='111@qq.com').all()

3. 自定义筛选条件
session.query(Users).filter(text("id>:vid or name=:vname")).params(vid=1, vname='xwx').all()
session.query(Users).filter(text("id>:vid and name=:vname")).params(vid=1, vname='xwx').all()

4. 自定义SQL语句
res = session.query(Users).from_statement(text("SELECT * FROM users where name='xwx'")).all()
res = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='xwx').all()

5. 二次筛选
session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='xwx'))).all()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3.6 条件连接

1. and
session.query(Users).filter(Users.id > 1, Users.name == 'lqz').all()

2. between
session.query(Users).filter(Users.id.between(7, 9)).all()

3. in_
session.query(Users).filter(Users.id.in_([7, 8, 9])).all()

4. ~ 取反
session.query(Users).filter(~Users.id.in_([1, 3, 4])).all()

5. or_ 或
session.query(Users).filter(or_(Users.id > 2, Users.name == 'xwx')).all()

6. and_ 与
session.query(Users).filter(and_(Users.id > 3, Users.name == 'xwx')).all()

7. and_、or_ 连用
session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'xwx', Users.id > 3),
        Users.extra != ""
    ))

 
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

3.7 模糊查询

session.query(Users).filter(Users.name.like('xwx%')).all()
session.query(Users).filter(~Users.name.like('%xwx%')).all()
1
2

3.8 分页

session.query(Users)[1:2]
1

3.9 排序

1. 根据 age 降序排列（从大到小）desc()
session.query(Users).order_by(Users.age.desc()).all()

2. 根据 age 降序排列（从小到大）asc()
session.query(Users).order_by(Users.id.desc()).all()
session.query(Users).order_by(Users.id.asc()).all()

3. 第一个条件重复后，再按第二个条件升序排列
session.query(Users).order_by(Users.age.desc(), Users.id.asc()).all()
1
2
3
4
5
6
7
8
9

3.10 分组

1. 按照 name 分组，等同于 sql 语句: 'select * from user group by name;'
session.query(Users).group_by(Users.name).all()

2. 分组后取最大、小值、总和，等同于: 'select max(id),sum(id),min(id) from user group by name;'
session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id),
).group_by(Users.name).all()

3. haviing 筛选，等同于: 'select max(id),sum(id),min(id) from user where email like 33% group by name  having min(id) > 2;'
session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)
).filter(Users.email.like('1%')).group_by(Users.name).having(func.min(Users.id) > 2).all()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

3.11 连表操作

1. 先笛卡尔积再过滤，等同于: 'select * from person,favor where user.id= favor.nid;' 
session.query(Users, Favor).filter(Users.id == Favor.nid).all()

2. join 表，默认是 inner join，等同于: 'select * from person inner join favor on person.favor_id=favor.id;'
session.query(Person).join(Favor).all()

3. 左连接，添加isouter=True，没有右连接。'select * from person left join favor on person.favor_id=favor.id;'
session.query(Person).join(Favor, isouter=True).all()

4. 右连接，表反过来
session.query(Favor).join(Person, isouter=True).all()

5. 指定连接条件，类似于: 'select * from person left join favor on person.id=favor.id;'
session.query(Person).join(Favor, Person.id==Favor.id, isouter=True).all()
1
2
3
4
5
6
7
8
9
10
11
12
13
14

3.12 union 和 union all 的区别

UNION：用于合并两个或多个 SELECT 语句的结果集，'不包括重复行'，同时进行默认规则的排序
Union All：对两个结果集进行并集操作，'包括重复行'，不进行排序； 
1
2

1. union 
q1 = session.query(Users.id).filter(Users.id > 3)
q2 = session.query(Users.id).filter(Users.id < 6)
ret = q1.union(q2).all()
print(ret)

2. union all
q1 = session.query(Users.id).filter(Users.id > 3)
q2 = session.query(Users.id).filter(Users.id < 6)
ret = q1.union_all(q2).all()
1
2
3
4
5
6
7
8
9
10

4 一对多外键关系

• 表结构
  这里使用人与爱好表示例，假设一个人有一个爱好，爱好可以有多个人喜欢

import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship

# 实例化得到对象
Base = declarative_base()


class Hobby(Base):
    __tablename__ = 'hobby'
    id = Column(Integer, primary_key=True)
    caption = Column(String(50), default='篮球')

    def __repr__(self):
        return self.caption


class Person(Base):
    __tablename__ = 'person'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    # hobby指的是表名tablename而不是类名，uselist=False
    hobby_id = Column(Integer, ForeignKey("hobby.id"))

    # 该字段与数据库无关，不会新增表字段，只用于快速链表操作（基于对象的跨表查询、backref用于反向查询）
    hobby = relationship('Hobby', backref='pers')

    def __repr__(self):
        return self.name


def init_db():
    """
    根据类创建数据库表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:xwx668428@127.0.0.1:3306/flask_sql?charset=utf8mb4",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )

    Base.metadata.create_all(engine)


if __name__ == '__main__':
    # 创建所有表
    init_db()
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

4.1 一对多添加数据

1. 方式一
session.add(Hobby(caption='足球'))
session.add(Person(name='xwx', hobby_id=1))

2. 方式二，对象的方式
session.add(Person(name='xwx', hobby=Hobby(caption='橄榄球')))
1
2
3
4
5
6

4.2 基于对象查询

• 正向查询，根据人物查询爱好

res = session.query(Person).filter(Person.name == 'xwx').first()
print(res.hobby.caption)	# 外键字段在人物表这，直接点就行
1
2

• 反向查询，根据爱好查询人物

res = session.query(Hobby).filter_by(caption='足球').first()
print(res.pers)		# 由于设置了 relationship 的 backref 可以直接点，结果类似于列表
print(res.pers[0].id)
1
2
3

4.3 基于连表查询

'select * from person，hobby where person.hobby_id=hobby.id and person.name=xwx;'
res = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id, Person.name == 'xwx').all()
print(res)
print(res[0])
1
2
3
4

'select * from person inner join hobby on person.hobby_id=hobby.id where person.name=xwx;'
res = session.query(Person, Hobby).join(Hobby).filter(Person.name == 'xwx').all()
print(res)
1
2
3

5. 多对多外键关系

• 表模型

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship

Base = declarative_base()



class Boy2Girl(Base):
    __tablename__ = 'boy2girl'
    id = Column(Integer, primary_key=True, autoincrement=True)
    girl_id = Column(Integer, ForeignKey('girl.id'))
    boy_id = Column(Integer, ForeignKey('boy.id'))


class Girl(Base):
    __tablename__ = 'girl'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)

    def __repr__(self):
        return self.name


class Boy(Base):
    __tablename__ = 'boy'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(64), unique=True, nullable=False)

    # 与生成表结构无关，仅用于查询方便, 放在哪个单表中都可以
    # 方便快速查询，写了这个字段，相当于django 的 manytomany，快速使用基于对象的跨表查询
    girls = relationship('Girl', secondary='boy2girl', backref='boys')

    def __repr__(self):
        return self.name

...
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

5.1 多对多添加数据

1. 方式一，直接添加
session.add(Boy(name='B1'))
session.add(Girl(name='G1'))

# 第三张表必须要已有数据
session.add_all([Boy(name='B1'), Girl(name='G1')])

# 或者使用该方式
session.add_all([Boy2Girl(boy_id=1, girl_id=1), Boy2Girl(boy_id=1, girl_id=2)])

2. 方式二，使用外键
session.add(Boy(name='B1', girls=[Girl(name='G1'), Girl(name='G2')]))
1
2
3
4
5
6
7
8
9
10
11
12

5.2 基于对象跨表查询

• 正向查询

res = session.query(Boy).filter_by(name='B1').first()
print(res.girls)	# 点外键字段，值为列表
1
2

• 反向查询

res = session.query(Girl).filter_by(name='G1').first()
print(res.boys)		# 点 relationship 的 backref
1
2

6. flask-sqlalchemy、flask-migrate

• flask-sqlalchemy 是 flask 和 SQLAchemy 的管理者，可以帮助我们快速把 sqlalchemy 集成到 flask 中。
• flask-migrate ：原本 SQLAlchemy 不支持表字段增加删除。使用了 flask-migrate，以后增加了字段，删除字段只需要两条迁移命令就完成了

python3 manage.py db init 		初始化，只执行一次
python3 manage.py db migrate 	等同于 makemigartions
python3 manage.py db upgrade 	等同于 migrate
1
2
3

• 项目同名包/init.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

# 第一步：类实例化得到对象
db = SQLAlchemy()

from .models import *
from .views import account


def create_app():
    app = Flask(__name__)
    # 配置文件
    app.config.from_object('settings.DevelopmentConfig')

    # 第二步： 将db注册到app中
    db.init_app(app)

    # 注册蓝图
    app.register_blueprint(account.account)

    return app
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

• 项目同名文件/models.py

from . import db


# 第三步：把db导入，直接继承db.Model
class Users(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)

    # email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return '' % self.username

1
2
3
4
5
6
7
8
9
10
11
12
13
14

• 项目同名文件/views/account.py

from flask import Blueprint
from .. import db
from .. import models

account = Blueprint('account', __name__)

@account.route('/login')
def login():
    # 第四步：使用session,直接db.session即可
    db.session.add(models.Users(username='lqz', email='123'))
    db.session.query(models.Users).all()
    db.session.commit()  
    return 'login'
1
2
3
4
5
6
7
8
9
10
11
12
13

• 总结

# 第一步：导入 SQLAlchemy，实例化得到对象db
# 第二步：注册 db.init_app(app)
# 第三步：创建表模型，继承 db.Model
# 第四步：使用 session
1
2
3
4