本文整理汇总了Python中sqlalchemy.event.listen函数的典型用法代码示例。如果您正苦于以下问题:Python listen函数的具体用法?Python listen怎么用?Python listen使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了listen函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_parent_instance_child_class_apply_after
def test_parent_instance_child_class_apply_after(self):
l1 = Mock()
l2 = Mock()
event.listen(self.TargetElement, "event_one", l2)
factory = self.TargetFactory()
element = factory.create()
element.run_event(1)
event.listen(factory, "event_one", l1)
element.run_event(2)
element.run_event(3)
# c1 gets no events due to _JoinedListener
# fixing the "parent" at construction time.
# this can be changed to be "live" at the cost
# of performance.
eq_(
l1.mock_calls, []
)
eq_(
l2.mock_calls,
[call(element, 1), call(element, 2), call(element, 3)]
)
开发者ID:domenkozar,项目名称:sqlalchemy,代码行数:27,代码来源:test_events.py
示例2: register_signals
def register_signals(self, app):
"""Register the signals."""
before_record_index.connect(inject_provisional_community)
if app.config['COMMUNITIES_OAI_ENABLED']:
listen(Community, 'after_insert', create_oaipmh_set)
listen(Community, 'after_delete', destroy_oaipmh_set)
inclusion_request_created.connect(new_request)
开发者ID:inveniosoftware,项目名称:invenio-communities,代码行数:7,代码来源:ext.py
示例3: init_sqlalchemy
def init_sqlalchemy(self, scheme, connection):
try:
import sqlalchemy
from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import scoped_session, sessionmaker
from torweb.db import CacheQuery
import _mysql_exceptions
from sqlalchemy import event
from sqlalchemy.exc import DisconnectionError
def my_on_checkout(dbapi_conn, connection_rec, connection_proxy):
try:
dbapi_conn.cursor().execute('select now()')
except _mysql_exceptions.OperationalError:
raise DisconnectionError
engine = create_engine(
connection,
convert_unicode=True,
encoding="utf-8",
pool_recycle=3600*7,
#echo_pool=True,
echo=False,
)
event.listen(engine, 'checkout', my_on_checkout)
metadata = MetaData(bind=engine)
session = scoped_session(sessionmaker(bind=engine, query_cls=CacheQuery))
sqlalchemy_sessions = [session]
DB_Session = sessionmaker(bind=engine)
return {"metadata":metadata, "session":session, "sqlalchemy_sessions":sqlalchemy_sessions}
#setattr(self.app, 'metadata', metadata)
#setattr(self.app, scheme.get('sqlalchemy', 'session'), session)
#setattr(self.app, 'sqlalchemy_sessions', sqlalchemy_sessions)
except Exception as e:
print e
开发者ID:finalbattle,项目名称:torweb,代码行数:34,代码来源:config.py
示例4: test_all_events
def test_all_events(self):
canary = []
def before_attach(obj, parent):
canary.append("%s->%s" % (obj.__class__.__name__, parent.__class__.__name__))
def after_attach(obj, parent):
canary.append("%s->%s" % (obj.__class__.__name__, parent))
event.listen(schema.SchemaItem, "before_parent_attach", before_attach)
event.listen(schema.SchemaItem, "after_parent_attach", after_attach)
m = MetaData()
t1 = Table('t1', m,
Column('id', Integer, Sequence('foo_id'), primary_key=True),
Column('bar', String, ForeignKey('t2.id'))
)
t2 = Table('t2', m,
Column('id', Integer, primary_key=True),
)
eq_(
canary,
['Sequence->Column', 'Sequence->id', 'ForeignKey->Column',
'ForeignKey->bar', 'Table->MetaData',
'PrimaryKeyConstraint->Table', 'PrimaryKeyConstraint->t1',
'Column->Table', 'Column->t1', 'Column->Table',
'Column->t1', 'ForeignKeyConstraint->Table',
'ForeignKeyConstraint->t1', 'Table->MetaData(bind=None)',
'Table->MetaData', 'PrimaryKeyConstraint->Table',
'PrimaryKeyConstraint->t2', 'Column->Table', 'Column->t2',
'Table->MetaData(bind=None)']
)
开发者ID:ContextLogic,项目名称:sqlalchemy,代码行数:32,代码来源:test_metadata.py
示例5: listen
def listen(self, cls, prp, handler):
if (cls, prp) not in self.handlers:
self.handlers[(cls, prp)] = list()
event.listen(class_mapper(cls)._props[prp],
'set', self.make_callback(cls, prp))
self.handlers[(cls, prp)].append(handler)
开发者ID:Mloc,项目名称:cms,代码行数:7,代码来源:smartmappedcollection.py
示例6: test_instance
def test_instance(self):
Target = self._fixture()
class Foo(object):
def __init__(self):
self.mock = Mock()
def evt(self, arg):
self.mock(arg)
f1 = Foo()
f2 = Foo()
event.listen(Target, "event_one", f1.evt)
event.listen(Target, "event_one", f2.evt)
t1 = Target()
t1.dispatch.event_one("x")
event.remove(Target, "event_one", f1.evt)
t1.dispatch.event_one("y")
eq_(f1.mock.mock_calls, [call("x")])
eq_(f2.mock.mock_calls, [call("x"), call("y")])
开发者ID:Callek,项目名称:sqlalchemy,代码行数:25,代码来源:test_events.py
示例7: test_listener_collection_removed_cleanup
def test_listener_collection_removed_cleanup(self):
from sqlalchemy.event import registry
Target = self._fixture()
m1 = Mock()
t1 = Target()
event.listen(t1, "event_one", m1)
key = (id(t1), "event_one", id(m1))
assert key in registry._key_to_collection
collection_ref = list(registry._key_to_collection[key])[0]
assert collection_ref in registry._collection_to_key
t1.dispatch.event_one("t1")
del t1
gc_collect()
assert key not in registry._key_to_collection
assert collection_ref not in registry._collection_to_key
开发者ID:Callek,项目名称:sqlalchemy,代码行数:25,代码来源:test_events.py
示例8: file_column
def file_column(column):
def set_event_listner(target, value, oldvalue, initiator):
if oldvalue:
full_path = FileProcess.fullPath(oldvalue)
if (os.path.isfile(full_path)):
os.unlink(full_path)
alchemy_event.listen(column, 'set', set_event_listner)
开发者ID:skoropets,项目名称:Bookmark-Board,代码行数:7,代码来源:model.py
示例9: __declare_last__
def __declare_last__(cls) -> None:
""" Executed by SQLAlchemy as part of mapper configuration
When the data changes, we want to reset the cursor position of target
instance, to allow proper streaming of data.
"""
event.listen(DBStoredFile.data, "set", handle_change_data)
开发者ID:Kotti,项目名称:Kotti,代码行数:7,代码来源:filedepot.py
示例10: initialize_sql_test
def initialize_sql_test(engine):
DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
DBSession.configure(bind=engine)
Base.metadata.bind = engine
Base.metadata.create_all(engine)
listen(sqlalchemy.orm.mapper, 'before_insert', make_uuid)
return [DBSession, Base]
开发者ID:azmikamis,项目名称:raggregate,代码行数:7,代码来源:__init__.py
示例11: define_temp_tables
def define_temp_tables(cls, metadata):
# cheat a bit, we should fix this with some dialect-level
# temp table fixture
if testing.against("oracle"):
kw = {
'prefixes': ["GLOBAL TEMPORARY"],
'oracle_on_commit': 'PRESERVE ROWS'
}
else:
kw = {
'prefixes': ["TEMPORARY"],
}
user_tmp = Table(
"user_tmp", metadata,
Column("id", sa.INT, primary_key=True),
Column('name', sa.VARCHAR(50)),
Column('foo', sa.INT),
sa.UniqueConstraint('name', name='user_tmp_uq'),
sa.Index("user_tmp_ix", "foo"),
**kw
)
if testing.requires.view_reflection.enabled and \
testing.requires.temporary_views.enabled:
event.listen(
user_tmp, "after_create",
DDL("create temporary view user_tmp_v as "
"select * from user_tmp")
)
event.listen(
user_tmp, "before_drop",
DDL("drop view user_tmp_v")
)
开发者ID:DanielPearl,项目名称:top_commodities,代码行数:33,代码来源:test_reflection.py
示例12: test_transactional_advanced
def test_transactional_advanced(self):
canary = []
def tracker(name):
def go(*args, **kw):
canary.append(name)
return go
engine = engines.testing_engine()
for name in ['begin', 'savepoint',
'rollback_savepoint', 'release_savepoint',
'rollback', 'begin_twophase',
'prepare_twophase', 'commit_twophase']:
event.listen(engine, '%s' % name, tracker(name))
conn = engine.connect()
trans = conn.begin()
trans2 = conn.begin_nested()
conn.execute(select([1]))
trans2.rollback()
trans2 = conn.begin_nested()
conn.execute(select([1]))
trans2.commit()
trans.rollback()
trans = conn.begin_twophase()
conn.execute(select([1]))
trans.prepare()
trans.commit()
eq_(canary, ['begin', 'savepoint',
'rollback_savepoint', 'savepoint', 'release_savepoint',
'rollback', 'begin_twophase',
'prepare_twophase', 'commit_twophase']
)
开发者ID:ContextLogic,项目名称:sqlalchemy,代码行数:35,代码来源:test_execute.py
示例13: test_transactional
def test_transactional(self):
canary = []
def tracker(name):
def go(conn, *args, **kw):
canary.append(name)
return go
engine = engines.testing_engine()
event.listen(engine, 'before_execute', tracker('execute'))
event.listen(engine, 'before_cursor_execute', tracker('cursor_execute'))
event.listen(engine, 'begin', tracker('begin'))
event.listen(engine, 'commit', tracker('commit'))
event.listen(engine, 'rollback', tracker('rollback'))
conn = engine.connect()
trans = conn.begin()
conn.execute(select([1]))
trans.rollback()
trans = conn.begin()
conn.execute(select([1]))
trans.commit()
eq_(canary, [
'begin', 'execute', 'cursor_execute', 'rollback',
'begin', 'execute', 'cursor_execute', 'commit',
])
开发者ID:ContextLogic,项目名称:sqlalchemy,代码行数:26,代码来源:test_execute.py
示例14: test_retval_flag
def test_retval_flag(self):
canary = []
def tracker(name):
def go(conn, *args, **kw):
canary.append(name)
return go
def execute(conn, clauseelement, multiparams, params):
canary.append('execute')
return clauseelement, multiparams, params
def cursor_execute(conn, cursor, statement,
parameters, context, executemany):
canary.append('cursor_execute')
return statement, parameters
engine = engines.testing_engine()
assert_raises(
tsa.exc.ArgumentError,
event.listen, engine, "begin", tracker("begin"), retval=True
)
event.listen(engine, "before_execute", execute, retval=True)
event.listen(engine, "before_cursor_execute", cursor_execute, retval=True)
engine.execute(select([1]))
eq_(
canary, ['execute', 'cursor_execute']
)
开发者ID:ContextLogic,项目名称:sqlalchemy,代码行数:29,代码来源:test_execute.py
示例15: register
def register(self):
event.listen(
self.engine, 'before_cursor_execute', self.before_cursor_execute
)
event.listen(
self.engine, 'after_cursor_execute', self.after_cursor_execute
)
开发者ID:tanny2015,项目名称:Blog_Flask,代码行数:7,代码来源:__init__.py
示例16: includeme
def includeme(config: Configurator) -> None:
""" Pyramid includeme hook.
:param config: app config
:type config: :class:`pyramid.config.Configurator`
"""
config.add_tween(
"kotti.filedepot.TweenFactory", over=tweens.MAIN, under=tweens.INGRESS
)
config.add_request_method(uploaded_file_response, name="uploaded_file_response")
config.add_request_method(uploaded_file_url, name="uploaded_file_url")
from kotti.events import objectevent_listeners
from kotti.events import ObjectInsert
from kotti.events import ObjectUpdate
from sqlalchemy.event import listen
from sqlalchemy.engine import Engine
listen(Engine, "engine_connect", adjust_for_engine)
configure_filedepot(config.get_settings())
# Update file metadata on change of blob data
objectevent_listeners[(ObjectInsert, DBStoredFile)].append(set_metadata)
objectevent_listeners[(ObjectUpdate, DBStoredFile)].append(set_metadata)
# depot's _SQLAMutationTracker._session_committed is executed on
# after_commit, that's too late for DBFileStorage to interact with the
# session
event.listen(DBSession, "before_commit", _SQLAMutationTracker._session_committed)
开发者ID:Kotti,项目名称:Kotti,代码行数:32,代码来源:filedepot.py
示例17: handle_error
def handle_error(engine, listener):
"""Add a handle_error listener for the given :class:`.Engine`.
This listener uses the SQLAlchemy
:meth:`sqlalchemy.event.ConnectionEvents.handle_error`
event, however augments the listener for pre-0.9.7 versions of SQLAlchemy
in order to support safe re-raise of the exception.
"""
if utils.sqla_100:
event.listen(engine, "handle_error", listener)
return
assert isinstance(engine, Engine), \
"engine argument must be an Engine instance, not a Connection"
if not utils.sqla_097:
_rework_handle_exception_for_events(engine)
engine._oslo_handle_error_events.append(listener)
_rework_connect_and_revalidate_for_events(engine)
if utils.sqla_097:
# ctx.engine added per
# https://bitbucket.org/zzzeek/sqlalchemy/issue/3266/
def wrap_listener(ctx):
if isinstance(ctx, engine_base.ExceptionContextImpl):
ctx.engine = ctx.connection.engine
return listener(ctx)
event.listen(engine, "handle_error", wrap_listener)
开发者ID:SvenDowideit,项目名称:clearlinux,代码行数:30,代码来源:handle_error.py
示例18: test_before_update_m2o
def test_before_update_m2o(self):
"""Expect normal many to one attribute load behavior
(should not get committed value)
from within public 'before_update' event"""
sess = self._mapper_setup()
Address, User = self.classes.Address, self.classes.User
def before_update(mapper, connection, target):
# if get committed is used to find target.user, then
# it will be still be u1 instead of u2
assert target.user.id == target.user_id == u2.id
from sqlalchemy import event
event.listen(Address, 'before_update', before_update)
a1 = Address(email_address='a1')
u1 = User(name='u1', addresses=[a1])
sess.add(u1)
u2 = User(name='u2')
sess.add(u2)
sess.commit()
sess.expunge_all()
# lookup an address and move it to the other user
a1 = sess.query(Address).get(a1.id)
# move address to another user's fk
assert a1.user_id == u1.id
a1.user_id = u2.id
sess.flush()
开发者ID:23andMe,项目名称:sqlalchemy,代码行数:32,代码来源:test_unitofworkv2.py
示例19: test_propagate
def test_propagate(self):
Target = self._fixture()
m1 = Mock()
t1 = Target()
t2 = Target()
event.listen(t1, "event_one", m1, propagate=True)
event.listen(t1, "event_two", m1, propagate=False)
t2.dispatch._update(t1.dispatch)
t1.dispatch.event_one("t1e1x")
t1.dispatch.event_two("t1e2x")
t2.dispatch.event_one("t2e1x")
t2.dispatch.event_two("t2e2x")
event.remove(t1, "event_one", m1)
event.remove(t1, "event_two", m1)
t1.dispatch.event_one("t1e1y")
t1.dispatch.event_two("t1e2y")
t2.dispatch.event_one("t2e1y")
t2.dispatch.event_two("t2e2y")
eq_(m1.mock_calls,
[call('t1e1x'), call('t1e2x'),
call('t2e1x')])
开发者ID:Callek,项目名称:sqlalchemy,代码行数:29,代码来源:test_events.py
示例20: test_accounting_commit_fails_delete
def test_accounting_commit_fails_delete(self):
User = self.classes.User
sess = create_session(autocommit=True)
fail = False
def fail_fn(*arg, **kw):
if fail:
raise Exception("commit fails")
event.listen(sess, "after_flush_postexec", fail_fn)
u1 = User(name='ed')
sess.add(u1)
sess.flush()
sess.delete(u1)
fail = True
assert_raises(
Exception,
sess.flush
)
fail = False
assert u1 in sess
assert u1 not in sess.deleted
sess.delete(u1)
sess.flush()
assert u1 not in sess
eq_(
sess.query(User.name).order_by(User.name).all(),
[]
)
开发者ID:LynYang,项目名称:sqlalchemy,代码行数:32,代码来源:test_transaction.py
注:本文中的sqlalchemy.event.listen函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论