SQLAlchemy

Question

में सिग्नल या ट्रिगर एसक्यूएलकेमी के पास Django की सिग्नल अवधारणा के समान कुछ है? असल में, जब मैं कुछ इकाई ऑब्जेक्ट्स को प्री-सेव या पोस्ट-सेव करता हूं तो मैं कुछ फ़ंक्शंस ट्रिगर करना चाहता हूं। धन्यवाद।

संपादित करें: मैं बस स्क्लाक्लेमी में django-संकेतों के बराबर चाहता हूं।

Answer 1

आपने स्पष्ट नहीं किया है, चाहे आप स्क्लेक्लेमी और डीजेगो को एकीकृत कर रहे हों, या आप बस स्क्लाक्लेमी में डीजेंगो-सिग्नल के बराबर चाहते हैं।

यदि आप चाहते हैं Django के बराबर post_save, pre_save, pre_delete आदि की तरह का संकेत, मैं तुम्हें पेज उल्लेख करता है,

sqlalchemy.orm.interfaces.MapperExtension

Answer 2

आप sqlalchemy.orm.SessionExtension पर विचार करने के साथ ही चाहते हो सकता है

यहाँ कुछ है कोड मैंने एक उदाहरण पर एक मालिक आईडी सेट करने के लिए एक साथ फेंक दिया और एक अपडेट_डेट सेट किया जो कि एक पिलोन ऐप में काम करता है। 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) 

Answer 3

यहाँ इस समस्या पर मेरी ले है, यह Louie का उपयोग करता संकेतों प्रेषण करने के लिए:

dispatch.py

""" 
Signals dispatching for SQLAlchemy mappers. 
""" 

import louie 
from sqlalchemy.orm.interfaces import MapperExtension 
import signals 


class LouieDispatcherExtension(MapperExtension): 
    """ 
    Dispatch signals using louie on insert, update and delete actions. 
    """ 

    def after_insert(self, mapper, connection, instance): 
     louie.send(signals.after_insert, instance.__class__, 
       instance=instance) 
     return super(LouieDispatcherExtension, self).after_insert(mapper, 
       connection, instance) 

    def after_delete(self, mapper, connection, instance): 
     louie.send(signals.after_delete, instance.__class__, 
       instance=instance) 
     return super(LouieDispatcherExtension, self).after_delete(mapper, 
       connection, instance) 

    def after_update(self, mapper, connection, instance): 
     louie.send(signals.after_update, instance.__class__, 
       instance=instance) 
     return super(LouieDispatcherExtension, self).after_update(mapper, 
       connection, instance) 

    def before_delete(self, mapper, connection, instance): 
     louie.send(signals.before_delete, instance.__class__, 
       instance=instance) 
     return super(LouieDispatcherExtension, self).before_delete(mapper, 
       connection, instance) 

    def before_insert(self, mapper, connection, instance): 
     louie.send(signals.before_insert, instance.__class__, 
       instance=instance) 
     return super(LouieDispatcherExtension, self).before_insert(mapper, 
       connection, instance) 

    def before_update(self, mapper, connection, instance): 
     louie.send(signals.before_update, instance.__class__, 
       instance=instance) 
     return super(LouieDispatcherExtension, self).before_update(mapper, 
       connection, instance) 

signals.py

from louie import Signal 


class after_delete(Signal): pass 
class after_insert(Signal): pass 
class after_update(Signal): pass 
class before_delete(Signal): pass 
class before_insert(Signal): pass 
class before_update(Signal): pass 

नमूना उपयोग:

class MyModel(DeclarativeBase): 

    __mapper_args__ = {"extension": LouieDispatcherExtension()} 

    ID = Column(Integer, primary_key=True) 
    name = Column(String(255)) 

def on_insert(instance): 
    print "inserted %s" % instance 

louie.connect(on_insert, signals.after_insert, MyModel) 

Answer 4

मुझे लगता है कि आप 'ओआरएम घटनाक्रम' की तलाश में हैं। आप यहाँ प्रलेखन पा सकते हैं:

http://docs.sqlalchemy.org/en/latest/orm/events.html

Answer 5

आप आंतरिक MapperExtension वर्ग का उपयोग कर सकते हैं:

class YourModel(db.Model): 

    class BaseExtension(MapperExtension): 

     def before_insert(self, mapper, connection, instance): 
      # do something here 

     def before_update(self, mapper, connection, instance): 
      # do something here 

    __mapper_args__ = { 'extension': BaseExtension() } 

    # ....