郑州网站建设工作_广州网络营销外包怎样_海淀搜索引擎优化seo_营销网站建设方案

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 mapperapp = 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 Flaskapp = 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)