Делает это, все еще перестали работать, если Вы изменяете это:
select new SomeObject { ... } as ISomeTable;
к этому:
select (ISomeTable) new SomeObject { ... };
?
Раз так (поскольку я вижу, что Вы подтвердили), возможно, это имеет отношение к тому, что интерфейсная реализация могла быть или классом или структурой? Проблема все еще появляется, если Вы бросаете к абстрактному классу, а не интерфейсу?
Вы не прояснили, интегрируете ли вы SQLAlchemy и Django, или вам просто нужен эквивалент сигналов django в SQLAlchemy.
Если вам нужен эквивалент сигналов Django, таких как post_save, pre_save, pre_delete и т. д., я бы отослал вас на страницу
Вы можете рассмотреть sqlalchemy .orm.SessionExtension также
Вот код, который я собрал, чтобы установить идентификатор владельца для экземпляра и установить update_date, который выполняет работу в приложении pylons. В классе OrmExt происходит вся магия. И init_model - это то место, где вы его подключаете.
import logging
import sqlalchemy as sa
from sqlalchemy import orm
from pylons import session
import datetime
log = logging.getLogger(__name__)
class ORMSecurityException(Exception):
'''
thrown for security violations in orm layer
'''
pass
def _get_current_user():
log.debug('getting current user from session...')
log.debug(session)
return session['user']
def _is_admin(user):
return False
def set_update_date(instance):
if hasattr(instance,'update_date'):
instance.update_date = datetime.datetime.now()
def set_owner(instance):
'''
if owner_id, run it through the rules
'''
log.info('set_owner')
if hasattr(instance, 'owner_id'):
log.info('instance.owner_id=%s' % instance.owner_id)
u = _get_current_user()
log.debug('user: %s' % u.email)
if not u:
#anonymous users can't save owned objects
raise ORMSecurityException()
if instance.owner_id==None:
#must be new object thus, owned by current user
log.info('setting owner on object %s for user: %s' % (instance.__class__.__name__,u.email))
instance.owner_id = u.id
elif instance.owner_id!=u.id and not _is_admin(u):
#if owner_id does not match user_id and user is not admin VIOLATION
raise ORMSecurityException()
else:
log.info('object is already owned by this user')
return #good to go
else:
log.info('%s is not an owned object' % instance.__class__.__name__)
return
def instance_policy(instance):
log.info('setting owner for %s' % instance.__class__.__name__)
set_owner(instance)
log.info('setting update_date for %s' % instance.__class__.__name__)
set_update_date(instance)
class ORMExt(orm.SessionExtension):
'''
attempt at managing ownership logic on objects
'''
def __init__(self,policy):
self._policy = policy
def before_flush(self,sqlsess,flush_context,instances):
'''
check all instances for owner_id==user.id
'''
try:
for instance in sqlsess.deleted:
try:
log.info('running policy for deleted %s' % instance.__class__.__name__)
self._policy(instance)
except Exception,ex:
log.error(ex)
raise ex
for instance in sqlsess.new:
try:
log.info('running policy for new %s' % instance.__class__.__name__)
self._policy(instance)
except Exception,ex:
log.error(ex)
raise ex
for instance in sqlsess.dirty:
try:
if sqlsess.is_modified(instance,include_collections=False,passive=True):
log.info('running policy for updated %s' % instance.__class__.__name__)
self._policy(instance)
except Exception, ex:
log.error(ex)
raise ex
except Exception,ex:
sqlsess.expunge_all()
raise ex
def init_model(engine):
"""Call me before using any of the tables or classes in the model"""
sm = orm.sessionmaker(autoflush=True, autocommit=True, bind=engine,extension=ORMExt(instance_policy))
meta.engine = engine
meta.Session = orm.scoped_session(sm)