<s> import copy <EOL> from zope . interface import implements <EOL> from twisted . words . protocols . jabber import jid <EOL> from wokkel . generic import parseXml , stripNamespace <EOL> from wokkel . pubsub import Subscription <EOL> from idavoll import error , iidavoll <EOL> class Storage : <EOL> implements ( iidavoll . IStorage ) <EOL> defaultConfig = { <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> } <EOL> } <EOL> def __init__ ( self , dbpool ) : <EOL> self . dbpool = dbpool <EOL> def getNode ( self , nodeIdentifier ) : <EOL> return self . dbpool . runInteraction ( self . _getNode , nodeIdentifier ) <EOL> def _getNode ( self , cursor , nodeIdentifier ) : <EOL> configuration = { } <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( nodeIdentifier , ) ) <EOL> row = cursor . fetchone ( ) <EOL> if not row : <EOL> raise error . NodeNotFound ( ) <EOL> if row . node_type == '<STR_LIT>' : <EOL> configuration = { <EOL> '<STR_LIT>' : row . persist_items , <EOL> '<STR_LIT>' : row . deliver_payloads , <EOL> '<STR_LIT>' : <EOL> row . send_last_published_item } <EOL> node = LeafNode ( nodeIdentifier , configuration ) <EOL> node . dbpool = self . dbpool <EOL> return node <EOL> elif row . node_type == '<STR_LIT>' : <EOL> configuration = { <EOL> '<STR_LIT>' : row . deliver_payloads , <EOL> '<STR_LIT>' : <EOL> row . send_last_published_item } <EOL> node = CollectionNode ( nodeIdentifier , configuration ) <EOL> node . dbpool = self . dbpool <EOL> return node <EOL> def getNodeIds ( self ) : <EOL> d = self . dbpool . runQuery ( """<STR_LIT>""" ) <EOL> d . addCallback ( lambda results : [ r [ <NUM_LIT:0> ] for r in results ] ) <EOL> return d <EOL> def createNode ( self , nodeIdentifier , owner , config ) : <EOL> return self . dbpool . runInteraction ( self . _createNode , nodeIdentifier , <EOL> owner , config ) <EOL> def _createNode ( self , cursor , nodeIdentifier , owner , config ) : <EOL> if config [ '<STR_LIT>' ] != '<STR_LIT>' : <EOL> raise error . NoCollections ( ) <EOL> owner = owner . userhost ( ) <EOL> try : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( nodeIdentifier , <EOL> config [ '<STR_LIT>' ] , <EOL> config [ '<STR_LIT>' ] , <EOL> config [ '<STR_LIT>' ] ) ) <EOL> except cursor . _pool . dbapi . IntegrityError : <EOL> raise error . NodeExists ( ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( owner , ) ) <EOL> if not cursor . fetchone ( ) : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( owner , ) ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( nodeIdentifier , owner ) ) <EOL> def deleteNode ( self , nodeIdentifier ) : <EOL> return self . dbpool . runInteraction ( self . _deleteNode , nodeIdentifier ) <EOL> def _deleteNode ( self , cursor , nodeIdentifier ) : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( nodeIdentifier , ) ) <EOL> if cursor . rowcount != <NUM_LIT:1> : <EOL> raise error . NodeNotFound ( ) <EOL> def getAffiliations ( self , entity ) : <EOL> d = self . dbpool . runQuery ( """<STR_LIT>""" , <EOL> ( entity . userhost ( ) , ) ) <EOL> d . addCallback ( lambda results : [ tuple ( r ) for r in results ] ) <EOL> return d <EOL> def getSubscriptions ( self , entity ) : <EOL> def toSubscriptions ( rows ) : <EOL> subscriptions = [ ] <EOL> for row in rows : <EOL> subscriber = jid . internJID ( '<STR_LIT>' % ( row . jid , <EOL> row . resource ) ) <EOL> subscription = Subscription ( row . node , subscriber , row . state ) <EOL> subscriptions . append ( subscription ) <EOL> return subscriptions <EOL> d = self . dbpool . runQuery ( """<STR_LIT>""" , <EOL> ( entity . userhost ( ) , ) ) <EOL> d . addCallback ( toSubscriptions ) <EOL> return d <EOL> def getDefaultConfiguration ( self , nodeType ) : <EOL> return self . defaultConfig [ nodeType ] <EOL> class Node : <EOL> implements ( iidavoll . INode ) <EOL> def __init__ ( self , nodeIdentifier , config ) : <EOL> self . nodeIdentifier = nodeIdentifier <EOL> self . _config = config <EOL> def _checkNodeExists ( self , cursor ) : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( self . nodeIdentifier , ) ) <EOL> if not cursor . fetchone ( ) : <EOL> raise error . NodeNotFound ( ) <EOL> def getType ( self ) : <EOL> return self . nodeType <EOL> def getConfiguration ( self ) : <EOL> return self . _config <EOL> def setConfiguration ( self , options ) : <EOL> config = copy . copy ( self . _config ) <EOL> for option in options : <EOL> if option in config : <EOL> config [ option ] = options [ option ] <EOL> d = self . dbpool . runInteraction ( self . _setConfiguration , config ) <EOL> d . addCallback ( self . _setCachedConfiguration , config ) <EOL> return d <EOL> def _setConfiguration ( self , cursor , config ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( config [ "<STR_LIT>" ] , <EOL> config [ "<STR_LIT>" ] , <EOL> config [ "<STR_LIT>" ] , <EOL> self . nodeIdentifier ) ) <EOL> def _setCachedConfiguration ( self , void , config ) : <EOL> self . _config = config <EOL> def getMetaData ( self ) : <EOL> config = copy . copy ( self . _config ) <EOL> config [ "<STR_LIT>" ] = self . nodeType <EOL> return config <EOL> def getAffiliation ( self , entity ) : <EOL> return self . dbpool . runInteraction ( self . _getAffiliation , entity ) <EOL> def _getAffiliation ( self , cursor , entity ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( self . nodeIdentifier , <EOL> entity . userhost ( ) ) ) <EOL> try : <EOL> return cursor . fetchone ( ) [ <NUM_LIT:0> ] <EOL> except TypeError : <EOL> return None <EOL> def getSubscription ( self , subscriber ) : <EOL> return self . dbpool . runInteraction ( self . _getSubscription , subscriber ) <EOL> def _getSubscription ( self , cursor , subscriber ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> userhost = subscriber . userhost ( ) <EOL> resource = subscriber . resource or '<STR_LIT>' <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( self . nodeIdentifier , <EOL> userhost , <EOL> resource ) ) <EOL> row = cursor . fetchone ( ) <EOL> if not row : <EOL> return None <EOL> else : <EOL> return Subscription ( self . nodeIdentifier , subscriber , row . state ) <EOL> def getSubscriptions ( self , state = None ) : <EOL> return self . dbpool . runInteraction ( self . _getSubscriptions , state ) <EOL> def _getSubscriptions ( self , cursor , state ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> query = """<STR_LIT>""" <EOL> values = [ self . nodeIdentifier ] <EOL> if state : <EOL> query += "<STR_LIT>" <EOL> values . append ( state ) <EOL> cursor . execute ( query , values ) <EOL> rows = cursor . fetchall ( ) <EOL> subscriptions = [ ] <EOL> for row in rows : <EOL> subscriber = jid . JID ( '<STR_LIT>' % ( row . jid , row . resource ) ) <EOL> options = { } <EOL> if row . subscription_type : <EOL> options [ '<STR_LIT>' ] = row . subscription_type <EOL> if row . subscription_depth : <EOL> options [ '<STR_LIT>' ] = row . subscription_depth <EOL> subscriptions . append ( Subscription ( self . nodeIdentifier , subscriber , <EOL> row . state , options ) ) <EOL> return subscriptions <EOL> def addSubscription ( self , subscriber , state , config ) : <EOL> return self . dbpool . runInteraction ( self . _addSubscription , subscriber , <EOL> state , config ) <EOL> def _addSubscription ( self , cursor , subscriber , state , config ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> userhost = subscriber . userhost ( ) <EOL> resource = subscriber . resource or '<STR_LIT>' <EOL> subscription_type = config . get ( '<STR_LIT>' ) <EOL> subscription_depth = config . get ( '<STR_LIT>' ) <EOL> try : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( userhost , ) ) <EOL> except cursor . _pool . dbapi . IntegrityError : <EOL> cursor . connection . rollback ( ) <EOL> try : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( resource , <EOL> state , <EOL> subscription_type , <EOL> subscription_depth , <EOL> self . nodeIdentifier , <EOL> userhost ) ) <EOL> except cursor . _pool . dbapi . IntegrityError : <EOL> raise error . SubscriptionExists ( ) <EOL> def removeSubscription ( self , subscriber ) : <EOL> return self . dbpool . runInteraction ( self . _removeSubscription , <EOL> subscriber ) <EOL> def _removeSubscription ( self , cursor , subscriber ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> userhost = subscriber . userhost ( ) <EOL> resource = subscriber . resource or '<STR_LIT>' <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( self . nodeIdentifier , <EOL> userhost , <EOL> resource ) ) <EOL> if cursor . rowcount != <NUM_LIT:1> : <EOL> raise error . NotSubscribed ( ) <EOL> return None <EOL> def isSubscribed ( self , entity ) : <EOL> return self . dbpool . runInteraction ( self . _isSubscribed , entity ) <EOL> def _isSubscribed ( self , cursor , entity ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( entity . userhost ( ) , <EOL> self . nodeIdentifier ) ) <EOL> return cursor . fetchone ( ) is not None <EOL> def getAffiliations ( self ) : <EOL> return self . dbpool . runInteraction ( self . _getAffiliations ) <EOL> def _getAffiliations ( self , cursor ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( self . nodeIdentifier , ) ) <EOL> result = cursor . fetchall ( ) <EOL> return [ ( jid . internJID ( r [ <NUM_LIT:0> ] ) , r [ <NUM_LIT:1> ] ) for r in result ] <EOL> class LeafNode ( Node ) : <EOL> implements ( iidavoll . ILeafNode ) <EOL> nodeType = '<STR_LIT>' <EOL> def storeItems ( self , items , publisher ) : <EOL> return self . dbpool . runInteraction ( self . _storeItems , items , publisher ) <EOL> def _storeItems ( self , cursor , items , publisher ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> for item in items : <EOL> self . _storeItem ( cursor , item , publisher ) <EOL> def _storeItem ( self , cursor , item , publisher ) : <EOL> data = item . toXml ( ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( publisher . full ( ) , <EOL> data , <EOL> self . nodeIdentifier , <EOL> item [ "<STR_LIT:id>" ] ) ) <EOL> if cursor . rowcount == <NUM_LIT:1> : <EOL> return <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( item [ "<STR_LIT:id>" ] , <EOL> publisher . full ( ) , <EOL> data , <EOL> self . nodeIdentifier ) ) <EOL> def removeItems ( self , itemIdentifiers ) : <EOL> return self . dbpool . runInteraction ( self . _removeItems , itemIdentifiers ) <EOL> def _removeItems ( self , cursor , itemIdentifiers ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> deleted = [ ] <EOL> for itemIdentifier in itemIdentifiers : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( self . nodeIdentifier , <EOL> itemIdentifier ) ) <EOL> if cursor . rowcount : <EOL> deleted . append ( itemIdentifier ) <EOL> return deleted <EOL> def getItems ( self , maxItems = None ) : <EOL> return self . dbpool . runInteraction ( self . _getItems , maxItems ) <EOL> def _getItems ( self , cursor , maxItems ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> query = """<STR_LIT>""" <EOL> if maxItems : <EOL> cursor . execute ( query + "<STR_LIT>" , <EOL> ( self . nodeIdentifier , <EOL> maxItems ) ) <EOL> else : <EOL> cursor . execute ( query , ( self . nodeIdentifier , ) ) <EOL> result = cursor . fetchall ( ) <EOL> items = [ stripNamespace ( parseXml ( r [ <NUM_LIT:0> ] ) ) for r in result ] <EOL> return items <EOL> def getItemsById ( self , itemIdentifiers ) : <EOL> return self . dbpool . runInteraction ( self . _getItemsById , itemIdentifiers ) <EOL> def _getItemsById ( self , cursor , itemIdentifiers ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> items = [ ] <EOL> for itemIdentifier in itemIdentifiers : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( self . nodeIdentifier , <EOL> itemIdentifier ) ) <EOL> result = cursor . fetchone ( ) <EOL> if result : <EOL> items . append ( parseXml ( result [ <NUM_LIT:0> ] ) ) <EOL> return items <EOL> def purge ( self ) : <EOL> return self . dbpool . runInteraction ( self . _purge ) <EOL> def _purge ( self , cursor ) : <EOL> self . _checkNodeExists ( cursor ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( self . nodeIdentifier , ) ) <EOL> class CollectionNode ( Node ) : <EOL> nodeType = '<STR_LIT>' <EOL> class GatewayStorage ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , dbpool ) : <EOL> self . dbpool = dbpool <EOL> def _countCallbacks ( self , cursor , service , nodeIdentifier ) : <EOL> """<STR_LIT>""" <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( service . full ( ) , <EOL> nodeIdentifier ) ) <EOL> results = cursor . fetchall ( ) <EOL> return results [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> def addCallback ( self , service , nodeIdentifier , callback ) : <EOL> def interaction ( cursor ) : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( service . full ( ) , <EOL> nodeIdentifier , <EOL> callback ) ) <EOL> if cursor . fetchall ( ) : <EOL> return <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( service . full ( ) , <EOL> nodeIdentifier , <EOL> callback ) ) <EOL> return self . dbpool . runInteraction ( interaction ) <EOL> def removeCallback ( self , service , nodeIdentifier , callback ) : <EOL> def interaction ( cursor ) : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( service . full ( ) , <EOL> nodeIdentifier , <EOL> callback ) ) <EOL> if cursor . rowcount != <NUM_LIT:1> : <EOL> raise error . NotSubscribed ( ) <EOL> last = not self . _countCallbacks ( cursor , service , nodeIdentifier ) <EOL> return last <EOL> return self . dbpool . runInteraction ( interaction ) <EOL> def getCallbacks ( self , service , nodeIdentifier ) : <EOL> def interaction ( cursor ) : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( service . full ( ) , <EOL> nodeIdentifier ) ) <EOL> results = cursor . fetchall ( ) <EOL> if not results : <EOL> raise error . NoCallbacks ( ) <EOL> return [ result [ <NUM_LIT:0> ] for result in results ] <EOL> return self . dbpool . runInteraction ( interaction ) <EOL> def hasCallbacks ( self , service , nodeIdentifier ) : <EOL> def interaction ( cursor ) : <EOL> return bool ( self . _countCallbacks ( cursor , service , nodeIdentifier ) ) <EOL> return self . dbpool . runInteraction ( interaction ) </s>
