sqlalchemy钩子

1. 基本概念

• Sqlalchemy: python Orm框架
• Session: sqlalchemy通过session和数据库进行交互, 用法是定义Session类(链接数据库)，实例session对象
• Events: Session和数据库交互过程中发生的各种事件, sqlalchemy包含一个事件API，该API发布了很多钩子; e.g.create transaction; flush; commit etc.
• State: 发生不同的SessionEvents之后的Session状态

2. 基本实现

• session实现: 定义Session类，然后实例化一个对象

>>> from sqlalchemy.orm import sessionmaker >>> Session = sessionmaker(bind=engine) >>> session = Session() # engine: 引擎，定义了链接的数据库 

• events实现: 事件注册，定义参数

from sqlalchemy.event import listen from sqlalchemy.pool import Pool def my_on_connect(dbapi_con, connection_record): print("New DBAPI connection:", dbapi_con) listen(Pool, 'connect', my_on_connect) 

3. Event

• Event Registration(事件注册): 通过api的方法listen()或者lister_for()装饰器来监听事件的发生。这两个方法接受三个参数：监听目标，拦截事件的标识符，自定义的监听的方法。位置参数和关键字参数只支持特殊类型的事件，可以指定事件函数的接口，或者提供关于给定事件的说明。事件的名字和一个相关监听器的标识参数，该参数是一个类的绑定的方法，例如PoolEvents.connect()的文档说该事件的名字为’connect’，并且用户自定义的监听方法应该接受两个位置参数。

from sqlalchemy.event import listen from sqlalchemy.pool import Pool def my_on_connect(dbapi_con, connection_record): print("New DBAPI connection:", dbapi_con) listen(Pool, 'connect', my_on_connect) # or from sqlalchemy.event import listens_for from sqlalchemy.pool import Pool @listens_for(Pool, "connect") def my_on_connect(dbapi_con, connection_record): print("New DBAPI connection:", dbapi_con) 

• Named Argument Styles(参数类型): 有许多种参数类型能被监听函数接收。PoolEvents.connect()接收dbapi_connection 和 connection_record两个参数，我们通过name来设置接收参数，通过建立接收器函数来接收关键字参数，通过name=True来决定接收listen还是listens_for

from sqlalchemy.event import listens_for from sqlalchemy.pool import Pool @listens_for(Pool, "connect", named=True) def my_on_connect(kw): print("New DBAPI connection:", kw['dbapi_connection']) # or from sqlalchemy.event import listens_for from sqlalchemy.pool import Pool @listens_for(Pool, "connect", named=True) def my_on_connect(dbapi_connection, kw): print("New DBAPI connection:", dbapi_connection) print("Connection record:", kw['connection_record']) 

• Targets （监听目标）: listen接收的目标可以是类，也可以是对象。

from sqlalchemy.event import listen from sqlalchemy.pool import Pool, QueuePool from sqlalchemy import create_engine from sqlalchemy.engine import Engine import psycopg2 def connect(): return psycopg2.connect(username='ed', host='127.0.0.1', dbname='test') my_pool = QueuePool(connect) my_engine = create_engine('postgresql://ed@localhost/test') # associate listener with all instances of Pool listen(Pool, 'connect', my_on_connect) # associate listener with all instances of Pool # via the Engine class listen(Engine, 'connect', my_on_connect) # associate listener with my_pool listen(my_pool, 'connect', my_on_connect) # associate listener with my_engine.pool listen(my_engine, 'connect', my_on_connect) 

• Modifiers （修改标识） : 一些监听器允许通过修改标识，通过retval=True来使用自定义监听函数的返回值。

def validate_phone(target, value, oldvalue, initiator): """Strip non-numeric characters from a phone number""" return re.sub(r'\D', '', value) # setup listener on UserContact.phone attribute, instructing # it to use the return value listen(UserContact.phone, 'set', validate_phone, retval=True) 

• API Reference :

1. sqlalchemy.event.listen(target, identifier, fn, *args, kw) 2. sqlalchemy.event.listens_for(target, identifier, *args, kw) # 移除监听事件 3. sqlalchemy.event.remove(target, identifier, fn) # 如果监听目标存在，则返回True 4. sqlalchemy.event.contains(target, identifier, fn) 

4. session_events

• 定义事件: 要明确事件的生命周期

e.g.:

from sqlalchemy import event from sqlalchemy.orm import sessionmaker def my_before_commit(session): print "before commit!" Session = sessionmaker() event.listen(Session, "before_commit", my_before_commit) 

• after_commit(session)

触发after_commit事件之后，Session就不是一个激活的事务了

e.g.:

from sqlalchemy import event # standard decorator style @event.listens_for(SomeSessionOrFactory, 'after_commit') def receive_after_commit(session): "listen for the 'after_commit' event" # ... (event handling logic) ... 

• after_flush(session, flush_context)

flush_context: flush上下文, UOWTransaction对象，记录flush的细节

e.g.:

from sqlalchemy import event # standard decorator style @event.listens_for(SomeSessionOrFactory, 'after_flush') def receive_after_flush(session, flush_context): "listen for the 'after_flush' event" # ... (event handling logic) ... 

• after_flush_postexec(session, flush_context)

flush发生，并且 post-exec 发生之后。

session的最终状态，

e.g.:

from sqlalchemy import event # standard decorator style @event.listens_for(SomeSessionOrFactory, 'after_flush_postexec') def receive_after_flush_postexec(session, flush_context): "listen for the 'after_flush_postexec' event" # ... (event handling logic) ... 

• before_commit(session)

commit之前

e.g.:

from sqlalchemy import event # standard decorator style @event.listens_for(SomeSessionOrFactory, 'before_commit') def receive_before_commit(session): "listen for the 'before_commit' event" # ... (event handling logic) ... 

• before_flush(session, flush_context, instances)

instances: None，废弃了

e.g.:

from sqlalchemy import event # standard decorator style @event.listens_for(SomeSessionOrFactory, 'before_flush') def receive_before_flush(session, flush_context, instances): "listen for the 'before_flush' event" # ... (event handling logic) ... 

5. model hook 框架

• 流程

# 1. 导入Model # 2. register_events # 3. 创建钩子实体 # 4. HookCenter.before_flush # 5. EntityHook.before_flush # 6. UserHook.pre_flush # before_flush # 1. collect_entities(session, 'before_flush') # 1.1 get_hookers # session._hookers = {'User': 'UserHook'} # 1.2 遍历session的状态列表['new', 'deleted', 'dirty'] # 1.3 调用hooker.collect_entitites_before_flush() # 将当前对象添加到实体的pre_flush_entities字典中 # 2. call_hookers(session, 'before_flush') # 2.1 get_hookers # 2.2 调用实体的before_flush # 2.3 调用钩子的pre_flush