libervia-pubsub: sat_pubsub/pgsql

comparison sat_pubsub/pgsql_storage.py @ 341:b49f75a26156

backend, pgsql: implemented subscriptionsGet and subscriptionsSet

author	Goffi <goffi@goffi.org>
date	Sun, 20 Aug 2017 11:56:04 +0200
parents	57a3051ee435
children	8cf1be9572f8

comparison

equal deleted inserted replaced

-:567e486bce24
+:b49f75a26156
 @param nodeDbId(unicode): database ID
 """
 return self.dbpool.runInteraction(self._getNodeById, nodeDbId)
 def _getNodeById(self, cursor, nodeDbId):
 cursor.execute("""SELECT node_id,
 node,
 node_type,
 persist_items,
 return self._buildNode(row)
 def getNode(self, nodeIdentifier, pep, recipient=None):
 return self.dbpool.runInteraction(self._getNode, nodeIdentifier, pep, recipient)
 def _getNode(self, cursor, nodeIdentifier, pep, recipient):
 cursor.execute(*withPEP("""SELECT node_id,
 node,
 node_type,
 persist_items,
 d = self.dbpool.runQuery("""SELECT node from nodes WHERE pep is {}NULL"""
 .format("NOT " if pep else ""))
 d.addCallback(lambda results: [r[0] for r in results])
 return d
 def createNode(self, nodeIdentifier, owner, config, pep, recipient=None):
 return self.dbpool.runInteraction(self._createNode, nodeIdentifier,
 owner, config, pep, recipient)
 def _createNode(self, cursor, nodeIdentifier, owner, config, pep, recipient):
 if config['pubsub#node_type'] != 'leaf':
 raise error.NoCollections()
 raise error.NodeNotFound()
 def deleteNode(self, nodeIdentifier, pep, recipient=None):
 return self.dbpool.runInteraction(self._deleteNode, nodeIdentifier, pep, recipient)
 def _deleteNode(self, cursor, nodeIdentifier, pep, recipient):
 cursor.execute(*withPEP("""DELETE FROM nodes WHERE node=%s""",
 (nodeIdentifier,), pep, recipient))
 if cursor.rowcount != 1:
 args.append(nodeIdentifier)
 cursor.execute(*withPEP(' '.join(query), args, pep, recipient))
 rows = cursor.fetchall()
 return [tuple(r) for r in rows]
 def getSubscriptions(self, entity, pep, recipient=None):
 def toSubscriptions(rows):
 subscriptions = []
 for row in rows:
 WHERE jid=%s AND nodes.pep=%s""",
 (entity.userhost(), recipient.userhost() if pep else None))
 d.addCallback(toSubscriptions)
 return d
 def getDefaultConfiguration(self, nodeType):
 return self.defaultConfig[nodeType]
 def formatLastItems(self, result):
 last_items = []
 for pep_jid_s, node, data, item_access_model in result:
 pep_jid = jid.JID(pep_jid_s)
 item = generic.stripNamespace(parseXml(data))
 last_items.append((pep_jid, node, item, item_access_model))
 return last_items
 def getLastItems(self, entities, nodes, node_accesses, item_accesses, pep):
 """get last item for several nodes and entities in a single request"""
 if not entities or not nodes or not node_accesses or not item_accesses:
 raise ValueError("entities, nodes and accesses must not be empty")
 def __init__(self, nodeDbId, nodeIdentifier, config):
 self.nodeDbId = nodeDbId
 self.nodeIdentifier = nodeIdentifier
 self._config = config
 def _checkNodeExists(self, cursor):
 cursor.execute("""SELECT 1 as exist FROM nodes WHERE node_id=%s""",
 (self.nodeDbId,))
 if not cursor.fetchone():
 raise error.NodeNotFound()
 def getType(self):
 return self.nodeType
 def getOwners(self):
 d = self.dbpool.runQuery("""SELECT jid FROM nodes NATURAL JOIN affiliations NATURAL JOIN entities WHERE node_id=%s and affiliation='owner'""", (self.nodeDbId,))
 d.addCallback(lambda rows: [jid.JID(r[0]) for r in rows])
 return d
 def getConfiguration(self):
 return self._config
 def setConfiguration(self, options):
 config = copy.copy(self._config)
 for option in options:
 config[option] = options[option]
 d = self.dbpool.runInteraction(self._setConfiguration, config)
 d.addCallback(self._setCachedConfiguration, config)
 return d
 def _setConfiguration(self, cursor, config):
 self._checkNodeExists(cursor)
 cursor.execute("""UPDATE nodes SET persist_items=%s,
 deliver_payloads=%s,
 config[const.OPT_SEND_LAST_PUBLISHED_ITEM],
 config[const.OPT_ACCESS_MODEL],
 config[const.OPT_PUBLISH_MODEL],
 self.nodeDbId))
 def _setCachedConfiguration(self, void, config):
 self._config = config
 def getMetaData(self):
 config = copy.copy(self._config)
 config["pubsub#node_type"] = self.nodeType
 return config
 def getAffiliation(self, entity):
 return self.dbpool.runInteraction(self._getAffiliation, entity)
 def _getAffiliation(self, cursor, entity):
 self._checkNodeExists(cursor)
 cursor.execute("""SELECT affiliation FROM affiliations
 NATURAL JOIN nodes
 try:
 return cursor.fetchone()[0]
 except TypeError:
 return None
 def getAccessModel(self):
 return self._config[const.OPT_ACCESS_MODEL]
 def getSubscription(self, subscriber):
 return self.dbpool.runInteraction(self._getSubscription, subscriber)
 def _getSubscription(self, cursor, subscriber):
 self._checkNodeExists(cursor)
 userhost = subscriber.userhost()
 if not row:
 return None
 else:
 return Subscription(self.nodeIdentifier, subscriber, row[0])
 def getSubscriptions(self, state=None):
 return self.dbpool.runInteraction(self._getSubscriptions, state)
 def _getSubscriptions(self, cursor, state):
 self._checkNodeExists(cursor)
 query = """SELECT node, jid, resource, state,
 subscriptions.append(Subscription(row[0], subscriber,
 row[3], options))
 return subscriptions
 def addSubscription(self, subscriber, state, config):
 return self.dbpool.runInteraction(self._addSubscription, subscriber,
 state, config)
 def _addSubscription(self, cursor, subscriber, state, config):
 self._checkNodeExists(cursor)
 userhost = subscriber.userhost()
 subscription_depth,
 userhost))
 except cursor._pool.dbapi.IntegrityError:
 raise error.SubscriptionExists()
 def removeSubscription(self, subscriber):
 return self.dbpool.runInteraction(self._removeSubscription,
 subscriber)
 def _removeSubscription(self, cursor, subscriber):
 self._checkNodeExists(cursor)
 userhost = subscriber.userhost()
 if cursor.rowcount != 1:
 raise error.NotSubscribed()
 return None
+def setSubscriptions(self, subscriptions):
+return self.dbpool.runInteraction(self._setSubscriptions, subscriptions)
+def _setSubscriptions(self, cursor, subscriptions):
+self._checkNodeExists(cursor)
+entities = self.getOrCreateEntities(cursor, [s.subscriber for s in subscriptions])
+entities_map = {jid.JID(e.jid): e for e in entities}
+# then we construct values for subscriptions update according to entity_id we just got
+placeholders = ','.join(len(subscriptions) * ["%s"])
+values = []
+for subscription in subscriptions:
+entity_id = entities_map[subscription.subscriber].entity_id
+resource = subscription.subscriber.resource or u''
+values.append((self.nodeDbId, entity_id, resource, subscription.state, None, None))
+# we use upsert so new values are inserted and existing one updated. This feature is only available for PostgreSQL >= 9.5
+cursor.execute("INSERT INTO subscriptions(node_id, entity_id, resource, state, subscription_type, subscription_depth) VALUES " + placeholders + " ON CONFLICT (entity_id, resource, node_id) DO UPDATE SET state=EXCLUDED.state", [v for v in values])
 def isSubscribed(self, entity):
 return self.dbpool.runInteraction(self._isSubscribed, entity)
 def _isSubscribed(self, cursor, entity):
 self._checkNodeExists(cursor)
 cursor.execute("""SELECT 1 as bool FROM entities
 (entity.userhost(),
 self.nodeDbId))
 return cursor.fetchone() is not None
 def getAffiliations(self):
 return self.dbpool.runInteraction(self._getAffiliations)
 def _getAffiliations(self, cursor):
 self._checkNodeExists(cursor)
 cursor.execute("""SELECT jid, affiliation FROM nodes
 def getOrCreateEntities(self, cursor, entities_jids):
 """Get entity_id from entities in entities table
 Entities will be inserted it they don't exist
 @param entities_jid(list[jid.JID]): entities to get or create
-@return list[record(entity_jid,jid)]]: list of entity_id and jid (as plain string)
+@return list[record(entity_id,jid)]]: list of entity_id and jid (as plain string)
 both existing and inserted entities are returned
 """
 # cf. http://stackoverflow.com/a/35265559
 placeholders = ','.join(len(entities_jids) * ["(%s)"])
 query = (
 return cursor.fetchall()
 def setAffiliations(self, affiliations):
 return self.dbpool.runInteraction(self._setAffiliations, affiliations)
 def _setAffiliations(self, cursor, affiliations):
 self._checkNodeExists(cursor)
 entities = self.getOrCreateEntities(cursor, affiliations)
 map(values.extend, ((e.entity_id, affiliations[jid.JID(e.jid)], self.nodeDbId) for e in entities))
 # we use upsert so new values are inserted and existing one updated. This feature is only available for PostgreSQL >= 9.5
 cursor.execute("INSERT INTO affiliations(entity_id,affiliation,node_id) VALUES " + placeholders + " ON CONFLICT  (entity_id,node_id) DO UPDATE SET affiliation=EXCLUDED.affiliation", values)
 def deleteAffiliations(self, entities):
 return self.dbpool.runInteraction(self._deleteAffiliations, entities)
 def _deleteAffiliations(self, cursor, entities):
 """delete affiliations and subscriptions for this entity"""
 self._checkNodeExists(cursor)
 placeholders = ','.join(len(entities) * ["%s"])
 """
 def __init__(self, dbpool):
 self.dbpool = dbpool
 def _countCallbacks(self, cursor, service, nodeIdentifier):
 """
 Count number of callbacks registered for a node.
 """
 cursor.execute("""SELECT count(*) FROM callbacks
 (service.full(),
 nodeIdentifier))
 results = cursor.fetchall()
 return results[0][0]
 def addCallback(self, service, nodeIdentifier, callback):
 def interaction(cursor):
 cursor.execute("""SELECT 1 as bool FROM callbacks
 WHERE service=%s and node=%s and uri=%s""",
 (service.full(),
 nodeIdentifier,
 callback))
 return self.dbpool.runInteraction(interaction)
 def removeCallback(self, service, nodeIdentifier, callback):
 def interaction(cursor):
 cursor.execute("""DELETE FROM callbacks
 WHERE service=%s and node=%s and uri=%s""",
 (service.full(),
 return [result[0] for result in results]
 return self.dbpool.runInteraction(interaction)
 def hasCallbacks(self, service, nodeIdentifier):
 def interaction(cursor):
 return bool(self._countCallbacks(cursor, service, nodeIdentifier))
 return self.dbpool.runInteraction(interaction)

Mercurial > libervia-pubsub

comparison sat_pubsub/pgsql_storage.py @ 341:b49f75a26156