当前位置：首页 > article >正文

flask_sqlalchemy event监听查询事件

article 2025/1/12 6:03:13

flask_sqlalchemy event监听查询事件
在Flask-SQLAlchemy中，可以使用事件监听器来监控查询事件。这可以通过@listens_for(ModelClass, “event_name”)装饰器来实现，其中ModelClass是你想要监控的模型类，event_name是你想要监控的事件名称，例如’before_insert’、‘after_insert’、'before_update’等。

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
# 监听模型创建事件
@event.listens_for(db.Model, 'after_insert')
def after_user_insert(mapper, connection, target):
    print(f'User {target.name} created')

# 监听模型删除事件
@event.listens_for(User, 'after_delete')
def after_user_delete(mapper, connection, target):
    print(f'User {target.name} deleted')

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import event
from sqlalchemy.orm import mapper
 
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:tmp/test.db'
db = SQLAlchemy(app)
 
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    email = db.Column(db.String(120), unique=True)
 
@event.listens_for(User, 'before_insert')
def before_user_insert(mapper, connection, user):
    print("Before User Insert: ", user.username)
 
@event.listens_for(User, 'after_insert')
def after_user_insert(mapper, connection, user):
    print("After User Insert: ", user.username)
 
# 创建数据库表
db.create_all()
 
# 使用模型类添加用户
user = User(username='john_doe', email='john@example.com')
db.session.add(user)
db.session.commit()

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.event import listens_for
from flask import Flask
 
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:tmp/test.db'
db = SQLAlchemy(app)
 
class MyModel(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
 
    def __repr__(self):
        return f"MyModel({self.name})"
 
@listens_for(db.session, "after_attach")
def after_attach(session, instance):
    if isinstance(instance, MyModel):
        print(f"Querying MyModel: {instance}")
 
if __name__ == "__main__":
    db.drop_all()
    db.create_all()
 
    # 创建一个模型实例
    m = MyModel(name='test')
    db.session.add(m)
    db.session.commit()
 
    # 查询时会触发上面定义的事件
    print("Querying all MyModels:")
    for model in MyModel.query.all():
        print(model)