JiFu's Wiki
JiFu
2023-10-18
目录

Flask-sqlalchmey介绍

# What is Flask-SQLAlchemy

Flask-SQLAlchemy is a Flask microframework extension which adds support for the SQLAlchemy SQL toolkit/ORM.

# Example


from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SECRET_KEY'] = 'Fianna'
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://user:password@host:port/dbname'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)
class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, nullable=False, primary_key=True, autoincrement=True)
    name = db.Column(db.String(16), nullable=False, server_default='', unique=True)
    def __repr__(self):
        return '<Role %r>' % self.name
class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, nullable=False, primary_key=True, autoincrement=True)
    username = db.Column(db.String(32), nullable=False, unique=True, server_default='', index=True)
    role_id = db.Column(db.Integer, nullable=False, server_default='0')
    def __repr__(self):
        return '<User %r,Role id %r>' %(self.username,self.role_id)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# pip依赖包安装

pip install mysql-python pip install flask-sqlalchemy

# 数据库URL

数据库引擎 URL
MySQL mysql://username:password@hostname/database
Postgres postgresql://username:password@hostname/database
SQLite(Unix) sqlite:////absolute/path/database
SQLite(Windows) sqlite:///c:/absolute/path/database

在这些URL中，hostname表示MySQL服务器所在的主机，可以是本地主机（localhost），也可以是远程服务器。数据服务器上可以托管多个数据库，因此database表示要是使用的数据库名。如果数据库需要进行认证，username和password表示数据库用户和密码。

程序中使用的数据库URL必须保存到Flask配置对象config.py的SQLALCHEMY_DATABASE_URI键中。另外一个很有用的选项，即SQLALCHEMY_COMMIT_ON_TEARDOWN键，将其设置为True时，每次请求结束后都会自动提交数据库中的变动。

# 创建迁移仓库

定义好数据模型之后，就应该设计数据库了，不管你用mysql还是sqlite，flask-SQLAlchemy和flask-Migrate会根据数据模型自动创建数据库。 为了导出数据库迁移命令，flask-Migrate提供了一个MigrateCommand类，可附加在flask-script的manage对象上。设置manage.py。

...
from flask.ext.migrate import Migrate, MigrateCommand

...

migrate = Migrate(app=app, db=db)
manager.add_command('db', MigrateCommand)
1
2
3
4
5
6
7

# 使用一下命令创建数据库及迁移仓库

(flask)$ python manage.py db init
(flask)$ python manage.py db migrate -m "initial migration"
(flask)$ python manage.py db upgrade
1
2
3

注意事项，虽然flask-migrate提供了downgrade()函数，可以将改动删除。但是建议大家不要随便使用。如果你觉得数据库设计得有问题，建议你删除相关数据库设计，重新再来。

# 定义数据模型

和所有的应用一样，我们先设计简单的用户和权限模型。模型这个术语表示程序中使用的持久化实体。在 ORM 中,模型一般是一个 Python 类,类中的属性对应数据库表中的列。 示例 2-1 app/models.py: 定义 Role 和 User 模型

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)

    def __repr__(self):
        return '<Role %r>' % self.name

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    def __repr__(self):
        return '<User %r>' % self.username
1
2
3
4
5
6
7
8
9
10
11
12
13
14

类变量__tablename__定义在数据库中使用的表名。其余的类变量都是该模型的属性，被定义为db.Column类的示例。

db.Column类构造函数的第一个参数是数据库列和模型属性的类型。下表列出了一些最常用的列类型以及在模型中使用的Python类型。

# AQLAlchemy列类型

类型名 Python类型 说明
Integer int 普通整数，一般是32位
SmallInteger int 取值范围小的整数，一般是16位
BigInteger int或long 不限制精度的整数
Float float 浮点数
Numeric decimal.Decimal 定点数
String str 变长字符串
Text str 变长字符串，对较长或不限长度的字符串做了优化
Unicode unicode 变长Unicode字符串
UnicodeText unicode 变长Unicode字符串,对较长或不限长度的字符床做了优化
Boolean bool 布尔值
Date datetime.date 日期
Time datetime.time 时间
DateTime datetime.datetime 日期和时间
Interval datetime.timedeta 时间间隔
Enum str 一组字符串
PickleType 任何Python对象 自动化使用Pickle序列化
LargeBinary str 二进制文件

# SQLAlchemy列选项

选项名 说明
primary_key 如果设为True，这列就是表的主键
unique 如果设为True，这列不允许出现重复
index 如果设为True，为这列创建索引，提升查询效率
nullable 如果设为True，这列允许使用空值；如果设为False，这列不允许使用空值
default 为这列定义默认值

# 关系

app/models.py 一对多关系模型

class Role(db.Model):
    # ...
    users = db.relationship('User', backref='role', lazy='dynamic')

class User(UserMixin, db.Model):
    # ...
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
1
2
3
4
5
6
7
8
9

# SQLAlchemy关系选项

选项名 说明
backref 在关系的另一个模型中添加反向引用
primaryjoin 明确指定两个模型之间使用的联结条件。只在模棱两可的关系中需要指定
lazy 指定如何加载相关记录。可选值有select（首次访问时按需加载）、immediate（源对象加载后加载）、joined（加载记录，但使用联结）、subquery（立即加载，但使用子查询），noload（永不加载）和dynamic（不加载记录，但提供加载记录的查询）
uselist 如果设为False，不使用列表，而使用标量值
order_by 指定关系中记录的排序方式
secondary 指定多对多关系表的名字
secondaryjoin SQLAlchemy无法自行决定时，指定多对多关系中的二级联结条件

除 一对多 关系之外，还有几种其它的关系类型。一对一 关系可以用前面介绍的 一对多 关系表示，但调用db.relationship()时要把uselist设为False，把“多”变成“一”。 多对一 关系也可使用 一对多 表示，对调两个表即可，或者把外键和db.relationship()都放在“多”这一侧。最复杂的关系是 多对多 ，需要用到第三张表，这个表称为 关系表。

# SQLAlchemy查询过滤器

过滤器 说明
filter() 把过滤器添加到原查询上，返回一个新查询
filter_by() 把等值过滤器添加到原查询上，返回一个新查询
limit() 使用指定的值限制原查询返回的结果数量，返回一个新查询
offset() 偏移原查询返回的结果，返回一个新查询
order_by() 根据指定条件对原查询结果进行排序，返回一个新查询
group_by() 根据指定条件对原查询进行分组，返回一个新查询

在查询上应用指定的过滤器后，通过调用all()执行查询，以列表形式返回结果。除了all()，还有其它方法能触发查询执行。

# SQLAlchemy查询执行函数

方法 说明
all() 以列表形式返回查询的所有结果
first() 返回查询的第一个结果，如果没有结果，返回None
first_or_404() 返回查询的第一个结果，如果没有结果，则终止请求，返回404错误响应
get() 返回指定主键对应的行，如果没有对应的行，返回None
get_or_404() 返回指定主键对应的行，如果没有对应的行，则终止请求，返回404错误响应
count() 返回查询结果的数量
paginate() 返回一个Paginate对象，它包含指定范围内的结果。

# 操作实例

# 连接数据库

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://user:password@host:port/dbname'

# 创建所有表

>>> from hello import db,Role,User
>>> db.create_all()
1
2

# 删除所有表

>>> from hello import db,Role,User
>>> db.drop_all()
1
2

# 插入行

# 插入单行

>>> from hello import db,Role,User
>>> db.session.add(Role(name='Admin'))
>>> db.session.commit()
>>> db.session.add(Role(name='Moderator'))
>>> db.session.add(Role(name='User'))
>>> db.session.commit()
1
2
3
4
5
6

# 插入多行

>>> from hello import db,Role,User
>>> db.session.add_all([User(username='john',role_id=1),User(username='susan',role_id=3),User(username='david',role_id=3)])
>>> db.session.commit()
1
2
3

# 更新行

>>> from hello import db,Role,User
>>> admin = Role.query.filter_by(name='Admin').first()
>>> admin.name='Administrator'
>>> db.session.commit()
1
2
3
4

# 删除行

>>> from hello import db,Role,User
>>> mod = Role.query.filter_by(name='Moderator').first()
>>> db.session.delete(mod)
>>> db.session.commit()
1
2
3
4

# 查询表

# 查询表中全部数据

# 注意，此处的查询结果完全取决于代码示例中的
# def __repr__(self)
>>> from hello import db,Role,User
>>> Role.query.all()
[<Role u'Administrator'>, <Role u'User'>]
>>> User.query.all()
[<User u'john',Role id 1L>, <User u'susan',Role id 3L>, <User u'david',Role id 3L>]
1
2
3
4
5
6
7

# 按照一个条件过滤数据记录(where)

>>> from hello import db,Role,User
>>> Role.query.filter_by(name='Administrator').first()
<Role u'Administrator'>
>>> User.query.filter_by(role_id=3).all()
[<User u'susan',Role id 3L>, <User u'david',Role id 3L>]
>>> User.query.filter_by(role_id=3).first()
<User u'susan',Role id 3L>
1
2
3
4
5
6
7

# 按照两个条件过滤数据记录(where and)

>>> from hello import db,Role,User
>>> User.query.filter_by(role_id=3,username='susan').first()
<User u'susan',Role id 3L>
>>> User.query.filter_by(role_id=3,username='susan').all()
[<User u'susan',Role id 3L>]
1
2
3
4
5

# 聚合(count)

python hello.py shell
>>> from hello import db,Role,User
>>> User.query.filter_by(role_id=3,username='susan').count()
1L
>>> User.query.filter_by(role_id=3).count()
2L
>>> User.query.count()
3L
1
2
3
4
5
6
7
8

# 求和(sum)

python hello.py shell
>>> from hello import db,Role,User
>>> from sqlalchemy.sql import func
>>> User.query.with_entities(func.sum(User.id)).all()
[(Decimal('6'),)]
>>> User.query.with_entities(func.sum(User.role_id)).all()
[(Decimal('7'),)]
1
2
3
4
5
6
7

# 平均数(avg)

python
>>> from hello import db,Role,User
>>> from sqlalchemy.sql import func
>>> User.query.with_entities(func.avg(User.role_id)).all()
[(Decimal('2.3333'),)]
>>> User.query.with_entities(func.avg(User.id)).all()
[(Decimal('2.0000'),)]
1
2
3
4
5
6
7

# 排序(order by)

python
>>> from hello import db,Role,User
# 升序(asc)
>>> User.query.order_by(User.role_id).all()
[<User u'john',Role id 1L>, <User u'susan',Role id 3L>, <User u'david',Role id 3L>]
# 降序(desc)
>>> User.query.order_by(User.role_id.desc()).all()
[<User u'susan',Role id 3L>, <User u'david',Role id 3L>, <User u'john',Role id 1L>]
1
2
3
4
5
6
7
8

# 分组(group by)

python hello.py shell
>>> from hello import db,Role,User
>>> User.query.group_by(User.role_id).all()
[<User u'john',Role id 1L>, <User u'susan',Role id 3L>]
1
2
3
4

# 限制(limit)

python
>>> from hello import db,Role,User
>>> User.query.all()
[<User u'john',Role id 1L>, <User u'susan',Role id 3L>, <User u'david',Role id 3L>]
# limit 1
>>> User.query.limit(1).all()
[<User u'john',Role id 1L>]
# limit 2,1
>>> User.query.limit(1).offset(2).all()
[<User u'david',Role id 3L>]
>>> User.query.filter_by(role_id=3).all()
[<User u'susan',Role id 3L>, <User u'david',Role id 3L>]
# limit 1
>>> User.query.filter_by(role_id=3).limit(1).all()
[<User u'susan',Role id 3L>]
# limit 1,1
>>> User.query.filter_by(role_id=3).limit(1).offset(1).all()
[<User u'david',Role id 3L>]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 将Flask-SQLAlchemy的查询语句转换为SQL

python
>>> from hello import db,Role,User
>>> User.query.all()
[<User u'john',Role id 1L>, <User u'susan',Role id 3L>, <User u'david',Role id 3L>]
>>> str(User.query)
'SELECT users.id AS users_id, users.username AS users_username, users.role_id AS users_role_id \nFROM users'
>>> User.query.limit(1).all()
[<User u'john',Role id 1L>]
>>> str(User.query.limit(1))
'SELECT users.id AS users_id, users.username AS users_username, users.role_id AS users_role_id \nFROM users \n LIMIT %s'
>>> User.query.limit(1).offset(2).all()
[<User u'david',Role id 3L>]
>>> str(User.query.limit(1).offset(2))
'SELECT users.id AS users_id, users.username AS users_username, users.role_id AS users_role_id \nFROM users \n LIMIT %s, %s'
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 参考资料

-第七章 使用 Flask 扩展管理数据库 (opens new window) -使用flask-sqlalchemy玩转MySQL (opens new window)

上次更新: 2024/08/11, 01:59:03
Flask介绍
SQLAlchemy介绍

← SQLAlchemy介绍→

最近更新
01
WPS快捷键
05-21
02
Disable notification "to get future google chrome updates you'll need macos 10.13 or later" on mac
05-14
03
MacOS软件推荐
04-30
更多文章>
Theme by Vdoing | Copyright © 2019-2025 Ji Fu | MIT License