Flask-SQLAlchemy安装及设置
- SQLALchemy 实际上是对数据库的抽象,让开发者不用直接和 SQL 语句打交道,而是通过 Python 对象来操作数据库,在舍弃一些性能开销的同时,换来的是开发效率的较大提升
- SQLAlchemy是一个关系型数据库框架,它提供了高层的 ORM 和底层的原生数据库的操作。flask-sqlalchemy 是一个简化了 SQLAlchemy 操作的flask扩展
文档地址:https://flask-sqlalchemy.palletsprojects.com/en/3.0.x/
SQLAlchemy文档:https://docs.sqlalchemy.org/en/20/
安装
- 安装 flask-sqlalchemy
pip install flask-sqlalchemy
- 如果连接的是 mysql 数据库,需要安装
pymysql
pip install pymysql
数据库连接设置
- 在 Flask-SQLAlchemy 中,数据库使用URL指定,而且程序使用的数据库必须保存到Flask配置对象的 SQLALCHEMY_DATABASE_URI 键中
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test'
- 其他设置:
# 动态追踪修改设置,如未设置只会提示警告
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
#查询时会显示原始SQL语句
app.config['SQLALCHEMY_ECHO'] = True
- 配置完成需要去 MySQL 中创建项目所使用的数据库
$ mysql -uroot -pmysql
$ create database test charset utf8;
- 其他配置
SQLALCHEMY_DATABASE_URI 用于连接的数据库 URI 。例如:sqlite:tmp/test.dbmysql://username:password@server/db
SQLALCHEMY_BINDS 一个映射 binds 到连接 URI 的字典。更多 binds 的信息见用 Binds 操作多个数据库。
SQLALCHEMY_ECHO 如果设置为Ture, SQLAlchemy 会记录所有 发给 stderr 的语句,这对调试有用。(打印sql语句)
SQLALCHEMY_RECORD_QUERIES 可以用于显式地禁用或启用查询记录。查询记录 在调试或测试模式自动启用。更多信息见get_debug_queries()。
SQLALCHEMY_NATIVE_UNICODE 可以用于显式禁用原生 unicode 支持。当使用 不合适的指定无编码的数据库默认值时,这对于 一些数据库适配器是必须的(比如 Ubuntu 上 某些版本的 PostgreSQL )。
SQLALCHEMY_POOL_SIZE 数据库连接池的大小。默认是引擎默认值(通常 是 5 )
SQLALCHEMY_POOL_TIMEOUT 设定连接池的连接超时时间。默认是 10 。
SQLALCHEMY_POOL_RECYCLE 多少秒后自动回收连接。这对 MySQL 是必要的, 它默认移除闲置多于 8 小时的连接。注意如果 使用了 MySQL , Flask-SQLALchemy 自动设定 这个值为 2 小时。
连接其他数据库
完整连接 URI 列表请跳转到 SQLAlchemy 下面的文档 (supported database) 。这里给出一些常见的连接字符串。
- Postgres:
postgresql://scott:tiger@localhost/mydatabase
- MySQL:
mysql://scott:tiger@localhost/mydatabase
or
mysql+pymysql://scott:tiger@localhost/mydatabase
- Oracle:
oracle://scott:tiger@127.0.0.1:1521/sidname
- SQLite (注意开头的四个斜线):
sqlite:absolute/path/to/foo.db
常用的SQLAlchemy字段类型
常用的SQLAl1chemy列选项
常用的SQLAlchemy关系选项
数据库基本操作
- 在Flask-SQLAlchemy中,插入、修改、删除操作,均由数据库会话管理。
- 会话用 db.session 表示。在准备把数据写入数据库前,要先将数据添加到会话中然后调用 commit() 方法提交会话。
- 在 Flask-SQLAlchemy 中,查询操作是通过 query 对象操作数据。
- 最基本的查询是返回表中所有数据,可以通过过滤器进行更精确的数据库查询。
在视图函数中定义模型类
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
#设置连接数据库的URL
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@localhost/test'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
#查询时会显示原始SQL语句
app.config['SQLALCHEMY_ECHO'] = True
db = SQLAlchemy(app)
class Role(db.Model):
# 定义表名,数据库中的真实表名
__tablename__ = 'roles'
# 定义列对象,变量名是数据库字段名
id = db.Column(db.Integer, primary_key=True) # 整型主键会默认设置自增主键
name = db.Column(db.String(64), unique=True)
users = db.relationship('User', backref='role')
#repr()方法显示一个可读字符串,使用print打印对象时,会输出这里定义的内容
def __repr__(self):
return 'Role:%s'% self.name
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True, index=True)
email = db.Column(db.String(64),unique=True)
password = db.Column(db.String(64))
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
def __repr__(self):
return 'User:%s'%self.name
@app.route("/")
def index():
return "index page"
if __name__ == '__main__':
app.run(debug=True)
创建表:
db.create_all()
删除表
db.drop_all()
插入一条数据
ro1 = Role(name='admin')
db.session.add(ro1)
db.session.commit()
再次插入一条数据
ro2 = Role(name='user')
db.session.add(ro2)
db.session.commit()
一次插入多条数据
us1 = User(name='wang',email='wang@163.com',password='123456',role_id=ro1.id)
us2 = User(name='zhang',email='zhang@189.com',password='201512',role_id=ro2.id)
us3 = User(name='chen',email='chen@126.com',password='987654',role_id=ro2.id)
us4 = User(name='zhou',email='zhou@163.com',password='456789',role_id=ro1.id)
us5 = User(name='tang',email='tang@itheima.com',password='158104',role_id=ro2.id)
us6 = User(name='wu',email='wu@gmail.com',password='5623514',role_id=ro2.id)
us7 = User(name='qian',email='qian@gmail.com',password='1543567',role_id=ro1.id)
us8 = User(name='liu',email='liu@itheima.com',password='867322',role_id=ro1.id)
us9 = User(name='li',email='li@163.com',password='4526342',role_id=ro2.id)
us10 = User(name='sun',email='sun@163.com',password='235523',role_id=ro2.id)
db.session.add_all([us1,us2,us3,us4,us5,us6,us7,us8,us9,us10])
db.session.commit()
查询
常用的SQLAlchemy查询过滤器(用于过滤)
filter() # 把过滤器添加到原查询上,返回一个新查询
filter_by() # 把等值过滤器添加到原查询上,返回一个新查询
limit # 使用指定的值限定原查询返回的结果
offset() # 偏移原查询返回的结果,返回一个新查询
order_by() # 根据指定条件对原查询结果进行排序,返回一个新查询
group_by() # 根据指定条件对原查询结果进行分组,返回一个新查询
以上过滤条件可连用,如User.query.filter().offset().order_by().limit()
常用的SQLAlchemy查询执行器(用于取结果)
all() # 以列表形式返回查询的所有结果
first() # 返回查询的第一个结果,如果未查到,返回None
first_or_404() # 返回查询的第一个结果,如果未查到,返回404
get() # 返回指定主键对应的行,如不存在,返回None
get_or_404() # 返回指定主键对应的行,如不存在,返回404
count() # 返回查询结果的数量
paginate() # 返回一个Paginate对象,它包含指定范围内的结果
查询:filter_by精确查询
返回名字等于wang的所有人
User.query.filter_by(name='wang').all() # flask-sqlalchemy提供的写法,query无参数
user = db.session.query(User).filter_by(name='wang').all() # sqlalchemy提供的写法,query有参数
执行sql,就是where中的过滤条件
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
WHERE users.name = %(name_1)s
first()返回查询到的第一个对象
User.query.first()
sql,可以看到就是limit
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
LIMIT %(param_1)s
all()返回查询到的所有对象
User.query.all()
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
filter模糊查询,返回名字结尾字符为g的所有数据
User.query.filter(User.name.endswith('g')).all()
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
WHERE (users.name LIKE concat('%%', %(name_1)s))
注意concat中的’%%'是python字符串的格式化写法,如:
a= "%%%s%%"%("abc")
实际表示的是
'%abc%'
而%(name_1)s
也是格式化的写法,如:
print(“I’m %(name)s. I’m %(age)d” % {‘name’:‘Pythontab’, ‘age’:99})
也就是最后的sql会是
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
WHERE (users.name LIKE concat('%', 'g')) - 也就是以g结尾
get():参数为主键,如果主键不存在没有返回内容
User.query.get(1)
逻辑非,返回名字不等于wang的所有数据
User.query.filter(User.name!='wang').all()
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
WHERE users.name != %(name_1)s
not_ 相当于取反
from sqlalchemy import not_
User.query.filter(not_(User.name=='chen')).all()
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
WHERE users.name != %(name_1)s
本质上与逻辑非没有区别
逻辑与,需要导入and,返回and()条件满足的所有数据
from sqlalchemy import and_
User.query.filter(and_(User.name!='wang',User.email.endswith('163.com'))).all()
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
WHERE users.name != %(name_1)s AND (users.email LIKE concat('%%', %(email_1)s))
逻辑或,需要导入or_
from sqlalchemy import or_
User.query.filter(or_(User.name!='wang',User.email.endswith('163.com'))).all()
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users
WHERE users.name != %(name_1)s OR (users.email LIKE concat('%%', %(email_1)s))
分组统计group_by
flask-sqlalchemy无法实现
from sqlalchemy import func
# 统计每个角色的用户数,即用户表以role_id分组统计count即可
ss = db.session.query(User.role_id, func.count(User.role_id)).group_by(User.role_id).all()
如果需要对分组后的数据进行过滤使用having
ss = db.session.query(User.role_id, func.count(User.role_id)).group_by(User.role_id).having(func.count(User.role_id)>1).all()
执行sql如下:
SELECT users.role_id AS users_role_id, count(users.role_id) AS count_1 FROM users GROUP BY users.role_id HAVING count(users.role_id) > 1
查询数据后删除
user = User.query.first()
db.session.delete(user)
db.session.commit()
User.query.all()
更新数据
user = User.query.first()
user.name = 'dong'
db.session.commit()
User.query.first()
模型之间的关联
一对多
前边定义的Role和User中,一对多的关系核心代码如下
class Role(db.Model):
...
#关键代码
users = db.relationship('User', backref='role')
...
class User(db.Model):
...
# 外键,数据库中真实存在的字段
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
其中realtionship
描述了Role
和User
的关系,在一对多的关系中,定义在一
的一方(本例中是一个角色对应多个用户,因此定义在Role类中)。
第一个参数为对应参照的类"User",以字符串输入
第二个参数backref
是为类User
创建的新属性,这样当有user对象时,可以使用user.role
得到该用户对应的Role
对象;此外如果有个role对象,要获取对应的所有用户时直接使用role.users
即可得到。
还有第三个参数lazy
决定了什么时候SQLALchemy从数据库中加载数据(案例中未设置)
- 如果设置为子查询方式(subquery),则会在加载完Role对象后,就立即加载与其关联的对象,这样会让总查询数量减少,但如果返回的条目数量很多,就会比较慢,设置为 subquery 的话,role.users 返回所有数据列表
- 也可以设置为动态方式(dynamic),这样关联对象会在被使用的时候再进行加载,并且在返回前进行过滤,如果返回的对象数很多,或者未来会变得很多,那最好采用这种方式。设置为 dynamic 的话,role.users 返回查询对象,并没有做真正的查询,可以利用查询对象做其他逻辑,比如:先排序再返回结果
User类中的role_id
不可省略,该字段设置的是外键,关联到roles表的id,当使用user.role
时,会使用该字段去得到role对象的所有属性,否则通过user.role_id
只能得到id,而不能得到其他属性。
多对多
需求举例:学生网上选课,每个学生可以选择多门课程,同时每门课程也可以被多个学生选择,构成多对多关系。多对多关系描述有一个唯一的点就是:需要添加一张单独的表去记录两张表之间的对应关系,这张表是数据库真实存在的表。
# 中间表,注意中间表使用db.Table,而非继承自db.Model,该表未设置主键
tb_student_course = db.Table('tb_student_course',
db.Column('student_id', db.Integer, db.ForeignKey('students.id')),
db.Column('course_id', db.Integer, db.ForeignKey('courses.id'))
)
# 学生表
class Student(db.Model):
__tablename__ = "students"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True)
# 关系,secondary指定关联中间表。通过backref使Course类中有student属性
courses = db.relationship('Course', secondary=tb_student_course,
backref='student',
lazy='dynamic')
# 课程表
class Course(db.Model):
__tablename__ = "courses"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True)
关联查询示例(一对多):
角色和用户的关系是一对多的关系,一个角色可以有多个用户,一个用户只能属于一个角色。
查询已知角色的所有用户
#查询roles表id为1的角色
ro1 = Role.query.get(1)
#查询该角色的所有用户
ro1.users.all()
查询已知用户所属角色
#查询users表id为3的用户
us1 = User.query.get(3)
#查询用户属于什么角色
us1.role
SELECT roles.id AS roles_id, roles.name AS roles_name
FROM roles
WHERE roles.id = %(pk_1)s
以上两个是已知role_id
或者user_id
,因此未涉及join查询
关联查询示例(多对多)
查询某个学生选修了哪些课程
stu1 = Student.query.get(1)
cos = stu1.courses # Student中有属性courses定义了关系
for co in cos:
print(co.name)
执行sql
SELECT courses.id AS courses_id, courses.name AS courses_name FROM courses, tb_student_course WHERE 1 = tb_student_course.student_id AND courses.id = tb_student_course.course_id
查询某个课程有哪些学生选修
co1 = Course.query.get(1)
stu = co1.student # [<Student 1>, <Student 2>, <Student 3>] 学生列表
for s in stu:
print(s.name)
执行sql
SELECT students.id AS students_id, students.name AS students_name FROM students, tb_student_course WHERE 1 = tb_student_course.course_id AND students.id = tb_student_course.student_id
join查询
- 内连接 inner join
# 以下查询结果中只包含user表的字段,不含角色表字段,实践中用途不大。得到的是User对象列表
# 其中User.role_id==Role.id在设置的有外键时,可不指定
d = User.query.join(Role, User.role_id==Role.id).all()
# sql
SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.password AS users_password, users.role_id AS users_role_id
FROM users INNER JOIN roles ON users.role_id = roles.id
下边查询特定字段,使用with_entities
传递要查询的字段信息
# 查询用户角色不为空时,用户的角色名(本质是inner join)
s = User.query.join(Role, User.role_id==Role.id).with_entities(User.id, User.name, User.role_id, Role.name).all()
# 原生sqlalchemy的写法
db.session.query(User.id, User.name, User.role_id, Role.name).join(User, User.role_id==Role.id, isouter=True).all()
# 结果样例,列表嵌套元组
[(6, 'wu', 2, 'user'), (7, 'qian', 1, 'admin'), (8, 'liu', 1, 'admin'), (9, 'li', 2, 'user'), (10, 'sun', 2, 'user')]
# 执行sql如下
SELECT users.id AS users_id, users.name AS users_name, users.role_id AS users_role_id, roles.name AS roles_name
FROM users INNER JOIN roles ON users.role_id = roles.id
- 左连接
d = User.query.join(Role, User.role_id==Role.id, isouter=True).with_entities(User.id, User.name, User.role_id, Role.name).all()
# sql
SELECT users.id AS users_id, users.name AS users_name, users.role_id AS users_role_id, roles.name AS roles_name
FROM users LEFT OUTER JOIN roles ON users.role_id = roles.id
- 右连接
与左连接雷同,两表反过来即可
另外有User.query.outerjoin
接口表示外连接,但其本质还是join,源码如下:
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4It8ln1s-1676342688920)(/upload/2023/02/image-1676017098431.png)]
关联查询并非强制要求模型类中设置外键,只要可以通过某个字段进行关联起来,即可使用join查询,只需要在join中指定关联字段即可。
子查询
# 下边代码是在Role和User表各添加了deleted字段,表示是否删除,0未删除,1删除。
# 通过subquery设定子查询
abc = db.session.query(User.id.label('id'), User.name.label('username'), Role.name.label('rolename'),
Role.deleted.label('deleted'), func.IF(Role.deleted==0,1,0).label('snapshot')).\
join(Role, Role.user_id == User.id, isouter=True).subquery()
xx = db.session.query(abc.c.id, abc.c.username, func.sum(abc.c.snapshot)).group_by(abc.c.id).all()
执行sql
select userid, username, sum(snp) from (
SELECT users.id userid, users.name username, roles.name rolename,roles.deleted, if(roles.deleted=0,1,0) as snp
FROM users LEFT OUTER JOIN roles ON roles.user_id = users.id) s GROUP BY s.userid;
常用函数
通常在数据库中使用的聚合函数如sum、count,if函数等都在sqlalchemy.func
中
from sqlalchemy import func
# 聚合函数
db.session.query(User.role_id, func.count(User.role_id)).group_by(User.role_id).all()
# if 函数
db.session.query(User.id, func.IF(User.id > 4, 1, 0).label('xx')) # user id大于4,返回1,否则返回0
其中label用来设置别名