Flask-SQLAlchemy

鄢雅畅

2023-12-01

Flask-SQLAlchemy

# SQLAlchemy是一个基于Python实现的ORM框架，跟web框架无关，独立的
# django的orm，sqlalchemy（大而重），peewee（小而轻）
# 异步orm框架：GINO
# Python界没有一个特别好的微服务框架：nameko
# java：dubbo（阿里开源），springcloud
# go：grpc，go-zero，go-micro

pip install flask-sqlalchemy

1、基本配置

配置选项	说明
`SQLALCHEMY_DATABASE_URI`	连接数据库。示例：`mysql://username:password@host/post/db?charset=utf-8`
`SQLALCHEMY_BINDS`	一个将会绑定多种数据库的字典。更多详细信息请看官文绑定多种数据库.
`SQLALCHEMY_ECHO`	调试设置为true
`SQLALCHEMY_POOL_SIZE`	数据库池的大小，默认值为5。
`SQLALCHEMY_POOL_TIMEOUT`	连接超时时间
`SQLALCHEMY_POOL_RECYCLE`	自动回收连接的秒数。
`SQLALCHEMY_MAX_OVERFLOW`	控制在连接池达到最大值后可以创建的连接数。当这些额外的连接回收到连接池后将会被断开和抛弃。
`SQLALCHEMY_TRACK_MODIFICATIONS`	如果设置成 True (默认情况)，Flask-SQLAlchemy 将会追踪对象的修改并且发送信号。这需要额外的内存，如果不必要的可以禁用它。

操作数据库需要先创建一个db对象，通常写在exts.py文件里。

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

flask项目一般将数据库配置写入configs.py文件里面，配置在创建引擎前需写好，不要在程序运行时修改配置，如下。

HOST = '127.0.0.1'
PORT = '3306'
DATABASE = 'flask1'
USERNAME = 'root'
PASSWORD = '123456'

DB_URI = "mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset=utf8".format(username=USERNAME,password=PASSWORD, host=HOST,port=PORT, db=DATABASE)

SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = True

写完数据库配置后需要和app绑定，app.py文件里写flask应用的创建和蓝图的注册等等，如下：

from flask import Flask
import configs
from exts import db

app = Flask(__name__)
# 加载配置文件
app.config.from_object(configs)
# db绑定app
db.init_app(app)

2、SQLAlchemy的数据类型

下图显示了最常用的SQLAlchemy列的类型：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-udGPQuRg-1636029658793)(20161107010358428)]

下图显示了最常用的SQLAlchemy列选项：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FpFLjBKw-1636029658794)(20161107010602160)]

Flask-SQLAlchemy 要求每个模型都要定义主键，这一列经常命名为 id。

下图展示了SQLAlchemy提供的过滤器和查询函数：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-69Yh4qM7-1636029658796)(20161107022638126)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-XDEwUEgq-1636029658797)(20161107023259291)]

3、示例

# -*- coding:utf-8 -*-
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
 
app = Flask(__name__)
# url的格式为：数据库的协议：//用户名：密码@ip地址：端口号（默认可以不写）/数据库名
app.config["SQLALCHEMY_DATABASE_URI"] = "mysql://root:mysql@localhost/first_flask"
# 动态追踪数据库的修改. 性能不好. 且未来版本中会移除. 目前只是为了解决控制台的提示才写的
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
# 创建数据库的操作对象
db = SQLAlchemy(app)
 
 
class Role(db.Model):
 
    __tablename__ = "roles"
    id = db.Column(db.Integer,primary_key=True)
    name = db.Column(db.String(16),unique=True)
    # 给Role类创建一个uses属性，关联users表。
    # backref是反向的给User类创建一个role属性，关联roles表。这是flask特殊的属性。
    users = db.relationship('User',backref="role")
    # 相当于__str__方法。
    def __repr__(self):
        return "Role: %s %s" % (self.id,self.name)
 
 
class User(db.Model):
    # 给表重新定义一个名称，默认名称是类名的小写，比如该类默认的表名是user。
    __tablename__ = "users"
    id = db.Column(db.Integer,primary_key=True)
    name = db.Column(db.String(16),unique=True)
    email = db.Column(db.String(32),unique=True)
    password = db.Column(db.String(16))
    # 创建一个外键，和django不一样。flask需要指定具体的字段创建外键，不能根据类名创建外键
    role_id = db.Column(db.Integer,db.ForeignKey("roles.id"))
 
    def __repr__(self):
        return "User: %s %s %s %s" % (self.id,self.name,self.password,self.role_id)
 
 
@app.route('/')
def hello_world():
    return 'Hello World!'
 
if __name__ == '__main__':
    # 删除所有的表
    db.drop_all()
    # 创建表
    db.create_all()
 
    ro1 = Role(name = "admin")
    # 先将ro1对象添加到会话中，可以回滚。
    db.session.add(ro1)
    
    '''
        在执行 db.session.commit() 提交到数据库出错时，需要执行数据库回滚，保证后续操作正常。
        try: 
            .
            .
        except Exception as e:
            db.session.rollback() # 执行数据库回滚
    '''
 
    try:
        ro2 = Role()
        ro2.name = 'user'
        db.session.add(ro2)
        # 最后插入完数据一定要提交
        db.session.commit()
    except Exception as e:
        db.session.rollback() 
        raise e
 
    try:
        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()
    except Exception as e:
        db.session.rollback()
        raise e
 
    app.run(debug=True)

4、一对多关系

# 一对多关系：一个Hobby可以有多个人喜欢，关联字段写在多的一方，Person

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


class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    # hobby指的是tablename而不是类名，跟hobby表的id字段建立外键关系
    hobby_id = Column(Integer, ForeignKey("hobby.id"))

    # 跟数据库无关，不会新增字段，只用于快速链表操作
    # 类名，backref用于反向查询
    hobby = relationship('Hobby', backref='persons')

5、多对多关系

# 多对多关系：男孩和女孩约会，一个男孩可以约多个女孩，一个女孩可以约多个男孩

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)


class Boy(Base):
    __tablename__ = 'boy'

    id = Column(Integer, primary_key=True, autoincrement=True) # autoincrement自增
    name = Column(String(64), unique=True, nullable=False)

    # 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
    girls = relationship('Girl', secondary='boy2girl', backref='boys')

6、SQLAlchemy的增删改

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from models import Users, Person, Hobby, Girl, Boy, Boy2Girl

# 单表的基本增删查改

# 第一步：得到engine对象
engine = create_engine("mysql+pymysql://root:111@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5)

# 第二步：得到Session对象，当成一个类
Session = sessionmaker(bind=engine)

# 第三步：创建session对象
# 每次执行数据库操作时，都需要创建一个session
session = Session()

# 第四步：以后使用session来操作数据

# #### 1增单个
# 先创建出一个user对象
# lqz = Users(name='lqz', email='33@qq.com')
# # 把对象增加到数据库中
# session.add(lqz)  # 只能同时增一个
# # 提交事务
# session.commit()
# # 把连接放回到池中
# session.close()


##### 2 同时增多个
# lqz = Users(name='lqz1', email='333@qq.com')
# egon = Users(name='egon', email='343@qq.com')
#
# lyf=Girl(name='刘亦菲')
# # 把对象增加到数据库中
# session.add_all([lqz,egon,lyf])
# # 提交事务
# session.commit()
# # 把连接放回到池中
# session.close()

##### 3 基本的查（查是最多的，现在先讲简单的）

# lqz=session.query(Users).filter_by(name='lqz').first()  # 查一个，返回Users对象
# lqz=session.query(Users).filter_by(name='lqz').all()      # 查所有,返回列表
# print(lqz)


#### 4 删除（查完再删）
# res=session.query(Users).filter_by(name='lqz').delete()
# session.commit()
# session.close()


### 5 修改
# res=session.query(Users).filter_by(name='lqz1').update({'name':'lqz_nb'})

# 类似于原来的F查询，把表中字段取出来使用
# synchronize_session=False  表示字符串相加
# res=session.query(Users).filter_by(name='lqz_nb').update({Users.name: Users.name + "099"}, synchronize_session=False)
# synchronize_session="evaluate" 表示数字相加（如果转不成数据，其实有错，直接设为0）
session.query(Users).filter_by(name='1').update({"id": Users.id + 10}, synchronize_session="evaluate")
session.commit()
session.close()

7、SQLAlchemy的查

1.以下的方法都是返回一个新的查询，需要配合执行器使用。
filter(): 过滤，功能比较强大，多表关联查询。
filter_by()：过滤，一般用在单表查询的过滤场景。
order_by()：排序。默认是升序，降序需要导包：from sqlalchemy import * 。然后引入desc方法。比如order_by(desc("email")).按照邮箱字母的降序排序。
group_by()：分组。

2.以下都是一些常用的执行器：配合上面的过滤器使用。
get()：获得id等于几的函数。
比如：查询id=1的对象。
get(1)，切记：括号里没有“id=”，直接传入id的数值就ok。因为该函数的功能就是查询主键等于几的对象。
all()：查询所有的数据。
first()：查询第一个数据。
count()：返回查询结果的数量。
paginate()：分页查询，返回一个分页对象。
paginate(参数1，参数2，参数3)=>参数1：当前是第几页；参数2：每页显示几条记录；参数3：是否要返回错误。
返回的分页对象有三个属性：items：获得查询的结果，pages：获得一共有多少页，page：获得当前页。

3.常用的逻辑符：
需要导入包才能用的有：from sqlalchemy import * 
not_、and_、or_ 还有上面说的排序desc。
常用的内置的有：in_：表示某个字段在什么范围之中。

4.其他关系的一些数据库查询：
endswith()：以什么结尾。
startswith()：以什么开头。
contains()：包含

示例

1. 查询所有用户数据
User.query.all()
 
2. 查询有多少个用户
User.query.count()
 
3. 查询第1个用户
User.query.first()
 
4. 查询id为4的用户[3种方式]
User.query.get(4)
User.query.filter_by(id=4).first()　　　　
User.query.filter(User.id==4).first()
 
filter:(类名.属性名==)
filter_by:(属性名=)
 
filter_by: 用于查询简单的列名，不支持比较运算符
filter比filter_by的功能更强大，支持比较运算符，支持or_、in_等语法。
 
5. 查询名字结尾字符为g的所有数据[开始/包含]
User.query.filter(User.name.endswith('g')).all()
User.query.filter(User.name.contains('g')).all()
 
6. 查询名字不等于wang的所有数据[2种方式]
from sqlalchemy import not_
PS：逻辑查询的格式：逻辑符_(类属性其他的一些判断)
User.query.filter(not_(User.name=='wang')).all()
User.query.filter(User.name!='wang').all()
 
7. 查询名字和邮箱都以 li 开头的所有数据[2种方式]
from sqlalchemy import and_
User.query.filter(and_(User.name.startswith('li'), User.email.startswith('li'))).all()
User.query.filter(User.name.startswith('li'), User.email.startswith('li')).all()
 
8. 查询password是 `123456` 或者 `email` 以 `itheima.com` 结尾的所有数据
from sqlalchemy import or_
User.query.filter(or_(User.password=='123456', User.email.endswith('itheima.com'))).all()
 
9. 查询id为 [1, 3, 5, 7, 9] 的用户列表
User.query.filter(User.id.in_([1, 3, 5, 7, 9])).all()
 
10. 查询name为liu的角色数据：关系引用
User.query.filter_by(name='liu').first().role.name
 
11. 查询所有用户数据，并以邮箱排序
User.query.order_by('email').all()  默认升序
User.query.order_by(desc('email')).all() 降序
 
12. 查询第2页的数据, 每页只显示3条数据
help(User.query.paginate)
pages = User.query.paginate(2, 3, False)
PS：三个参数: 1. 当前要查询的页数 2. 每页的数量 3. 是否要返回错误
pages.items # 获取查询的结果
pages.pages # 总页数
pages.page # 当前页数

8、Flask-SQLAlchemy使用

# 使用步骤
1 导入from flask_sqlalchemy import SQLAlchemy 实例化得到db对象
2 在app中注册
	db.init_app(app)
3 表模型继承 db.Model
4  session是db.session
	db.session使用即可
    
    
    
#存在问题
	1 表迁移麻烦
    2 不支持字段的动态修改

9、flask-migrate使用

# 像djagno一样，执行两条迁移命令，实现数据库的动态迁移
# pip install flask-migrate

# 使用步骤
第一步：from flask_migrate import Migrate, MigrateCommand
第二步：执行
Migrate(app, db)
manager.add_command('db', MigrateCommand)

python3 manage.py db init 初始化：只执行一次，生成一个migrations文件夹

# 以后直接新建表，新建字段，执行命令，就会自动同步
python3 manage.py db migrate 等同于 makemigartions
python3 manage.py db upgrade 等同于migrate

Flask-SQLAlchemy