python中的flask_sqlalchemy的使用及示例详解

目录

• @orm.reconstructor
• SQLAlchemy的回滚
• 关联其他模型
• 数据库基本操作
• 将数据添加到会话中示例：
• 在视图函数中定义模型类
• 常用的SQLAlchemy查询过滤器
• 常用的SQLAlchemy查询执行器
• 创建表：
• 删除表
• 插入一条数据
• 一次插入多条数据
• 查询:filter_by精确查询
• first()返回查询到的第一个对象
• all()返回查询到的所有对象
• filter模糊查询，返回名字结尾字符为g的所有数据。
• get()，参数为主键，如果主键不存在没有返回内容
• 逻辑非，返回名字不等于wang的所有数据。
• 逻辑与，需要导入and，返回and()条件满足的所有数据。
• 逻辑或，需要导入or_
• not_ 相当于取反
• 查询数据后删除
• 更新数据
• 使用update
• 关联查询示例：角色和用户的关系是一对多的关系，一个角色可以有多个用户，一个用户只能属于一个角色。

sqlalchemy创建实例的时候并不是像我们普通通过构造函数这样创建的方式来创建的， 而是通过元类的方式动态的创建实例对象。在python里面绝大多数的ORM都是通过元类的方式来创建对象，而不是通过普通实例化方式来创建的。那么通过元类的方式创建至少在sqlalchemy里面是不会通过构造函数来创建的。

如果我们就想在模型对象实例化的时候执行__init__，我们就需要用到sqlalchemy比较高级的机制，通过下面的官方文档我们知道是明确不会执行__init__的

但是可以通过装饰器来解决，如果我们在__init__打上这个装饰器，那么在模型对象实例化的时候，sqlalchemy就会去执行被打上装饰器这样的方法，

我们需要先导入orm

上图如果报错就使用下面的代码

或者

或者

db = SQLAlchemy(app)需要注意app对象能不能在当前.py文件中获取到

sqlalchemy会尝试为我们创建一个Base表，但是我们不需要他帮我们创建这个数据表，因为我们的Base是基类，是为了让所有的子类继承的，从而实现软删除（通过状态来标注数据是否可用，进而达到删除的目的）

只需要加入__abstract__=True

这样sqlalchemy就不会认为你要创建的是这样的表了

这里补充说明一下，类是可以多继承的。

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# *************************************
# author: jenrey
# create: 2019/1/27 22:29
# filename: user.py
# usage:测试SQLAlchemy
#
#
# *************************************
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import Column, Integer, String
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:root@localhost:3306/sqlalchemy'  # 用于连接数据库
app.config[
    'SQLALCHEMY_TRACK_MODIFICATIONS'] = True  # 如果设置成True(默认情况)，Flask-SQLAlchemy 将会追踪对象的修改并且发送信号。这需要额外的内存， 如果不必要的可以禁用它。如果你不显示的调用它，在最新版的运行环境下，会显示警告。
db = SQLAlchemy(app) # 注意这里要写app，因为没有通过请求访问的话是会报一个找不到上下文对象的错误的，而且不能采用上面是的三种方式。原因不详
class User(db.Model):
    # __tablename__ = 'Hehe'  # 指定表名，不写默认就是类名
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    age = Column(Integer, default=18)
    def __init__(self, id, name, age):
        self.id = id
        self.name = name
        self.age = age
    def __str__(self):  # print(实例对象)
        return "heihie"
    def __repr__(self):  # print(repr(实例对象))
        return "hehe"
db.init_app(app)
db.create_all()  # https://blog.csdn.net/JENREY/article/details/86607343
u1 = User(123, 'wangwu', 99) # 也可以不写def __init__(self)方法，传入值就需要指定u1 = User(id=123, name='wangwu')
db.session.add(u1)
db.session.commit()
if __name__ == '__main__':
    app.run()
    # pass

上面的代码可以看出我们是没有进行事务的操作的，但是SQLAlchemy天然的是支持事务的，所以我们在commit之前的代码都不会被真正的执行添加到数据库中的。但是上面代码还是有问题的，我们是没有进行数据的回滚（rollback）

为什么一定要执行db.session.rollback()，是因为如果我们的程序执行到db.session.commit()这里出现了错误，又没有执行rollback的话，那么不仅仅是这次的插入失败了，后续的其他操作SQLAlchemy也是会失败的。所以建议大家一定要回滚起来！

所以以后只要使用db.session.commit()就一定要用try except给包裹起来并在except里面执行db.session.rollback()

在Flask-SQLAlchemy中，插入、修改、删除操作，均由数据库会话管理。会话用db.session表示。在准备把数据写入数据库前，要先将数据添加到会话中然后调用commit()方法提交会话。

数据库会话是为了保证数据的一致性，避免因部分更新导致数据不一致。提交操作把会话对象全部写入数据库，如果写入过程发生错误，整个会话都会失效。

数据库会话也可以回滚，通过db.session.rollback()方法，实现会话提交数据前的状态。

在Flask-SQLAlchemy中，查询操作是通过query对象操作数据。最基本的查询是返回表中所有数据，可以通过过滤器进行更精确的数据库查询。

user = User(name='python')
db.session.add(user)
db.session.commit()

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
#设置连接数据库的URL
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/Flask_test'
#设置每次请求结束后会自动提交数据库中的改动
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
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)
    us = db.relationship('User', backref='role')
    #repr()方法显示一个可读字符串
    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)
    pswd = db.Column(db.String(64))
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    def __repr__(self):
        return 'User:%s'%self.name
if __name__ == '__main__':
    db.drop_all()
    db.create_all()
    ro1 = Role(name='admin')
    ro2 = Role(name='user')
    db.session.add_all([ro1,ro2])
    db.session.commit()
    us1 = User(name='wang',email='wang@163.com',pswd='123456',role_id=ro1.id)
    us2 = User(name='zhang',email='zhang@189.com',pswd='201512',role_id=ro2.id)
    us3 = User(name='chen',email='chen@126.com',pswd='987654',role_id=ro2.id)
    us4 = User(name='zhou',email='zhou@163.com',pswd='456789',role_id=ro1.id)
    db.session.add_all([us1,us2,us3,us4])
    db.session.commit()
    app.run(debug=True)

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

方法    说明
all()    以列表形式返回查询的所有结果
first()    返回查询的第一个结果，如果未查到，返回None
first_or_404()    返回查询的第一个结果，如果未查到，返回404
get()    返回指定主键对应的行，如不存在，返回None
get_or_404()    返回指定主键对应的行，如不存在，返回404
count()    返回查询结果的数量
paginate()    返回一个Paginate对象，它包含指定范围内的结果

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',pswd='123456',role_id=ro1.id)
us2 = User(name='zhang',email='zhang@189.com',pswd='201512',role_id=ro2.id)
us3 = User(name='chen',email='chen@126.com',pswd='987654',role_id=ro2.id)
us4 = User(name='zhou',email='zhou@163.com',pswd='456789',role_id=ro1.id)
db.session.add_all([us1,us2,us3,us4])
db.session.commit()

返回名字等于wang的所有人

User.query.filter_by(name='wang').all()

User.query.first()

User.query.all()

User.query.filter(User.name.endswith('g')).all()

User.query.get()

User.query.filter(User.name!='wang').all()

from sqlalchemy import and_
User.query.filter(and_(User.name!='wang',User.email.endswith('163.com'))).all()

from sqlalchemy import or_
User.query.filter(or_(User.name!='wang',User.email.endswith('163.com'))).all()

from sqlalchemy import not_
User.query.filter(not_(User.name=='chen')).all()

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()

User.query.filter_by(name='zhang').update({'name':'li'})

查询角色的所有用户：

#查询roles表id为1的角色
ro1 = Role.query.get(1)
#查询该角色的所有用户
ro1.us

查询用户所属角色：

​#查询users表id为3的用户
us1 = User.query.get(3)
#查询用户属于什么角色
us1.role

到此这篇关于python中的flask_sqlalchemy的使用及示例详解的文章就介绍到这了,更多相关python flask_sqlalchemy使用内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• python flask sqlalchemy连接数据库流程介绍
• Python flask sqlalchemy的简单使用及常用操作
• Python使用Flask-SQLAlchemy连接数据库操作示例
• Python利用flask sqlalchemy实现分页效果
• Python的Flask框架中使用Flask-SQLAlchemy管理数据库的教程
• 在Python程序和Flask框架中使用SQLAlchemy的教程
• Python的Flask框架中SQLAlchemy使用时的乱码问题解决