Mercurial > libervia-pubsub

comparison idavoll/pgsql_backend.py @ 59:0fa161c00ed9

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Use jid.JIDs everywhere in the backend. Fix subscription list retrieval in pgsql_backend, thereby actually making notifications work.
author Ralph Meijer <ralphm@ik.nu>
date Sat, 06 Nov 2004 17:07:47 +0000
parents 7c4dfef5d964
children 8cdbc27c467a
comparison
equal deleted inserted replaced
58:3e2e0040e3e0 59:0fa161c00ed9
36 self._check_node_exists(cursor, node_id) 36 self._check_node_exists(cursor, node_id)
37 cursor.execute("""SELECT affiliation FROM affiliations 37 cursor.execute("""SELECT affiliation FROM affiliations
38 JOIN nodes ON (node_id=nodes.id) 38 JOIN nodes ON (node_id=nodes.id)
39 JOIN entities ON (entity_id=entities.id) 39 JOIN entities ON (entity_id=entities.id)
40 WHERE node=%s AND jid=%s""", 40 WHERE node=%s AND jid=%s""",
41 (node_id.encode('utf8'), entity.encode('utf8'))) 41 (node_id.encode('utf8'),
42 entity.full().encode('utf8')))
42 43
43 try: 44 try:
44 return cursor.fetchone()[0] 45 return cursor.fetchone()[0]
45 except TypeError: 46 except TypeError:
46 return None 47 return None
48 def get_affiliation(self, node_id, entity): 49 def get_affiliation(self, node_id, entity):
49 return self.dbpool.runInteraction(self._get_affiliation, node_id, 50 return self.dbpool.runInteraction(self._get_affiliation, node_id,
50 entity) 51 entity)
51 52
52 def get_subscribers(self, node_id): 53 def get_subscribers(self, node_id):
53 self._check_node_exists(cursor, node_id) 54 d = self.dbpool.runInteraction(self._get_subscribers, node_id)
54 d = self.dbpool.runQuery("""SELECT jid, resource FROM subscriptions
55 JOIN nodes ON (node_id=nodes.id)
56 JOIN entities ON (entity_id=entities.id)
57 WHERE node=%s AND
58 subscription='subscribed'""",
59 (node_id.encode('utf8'),))
60 d.addCallback(self._convert_to_jids) 55 d.addCallback(self._convert_to_jids)
61 return d 56 return d
57
58 def _get_subscribers(self, cursor,node_id):
59 self._check_node_exists(cursor, node_id)
60 cursor.execute("""SELECT jid, resource FROM subscriptions
61 JOIN nodes ON (node_id=nodes.id)
62 JOIN entities ON (entity_id=entities.id)
63 WHERE node=%s AND
64 subscription='subscribed'""",
65 (node_id.encode('utf8'),))
66 return cursor.fetchall()
62 67
63 def _convert_to_jids(self, list): 68 def _convert_to_jids(self, list):
64 return [jid.JID("%s/%s" % (l[0], l[1])).full() for l in list] 69 return [jid.JID("%s/%s" % (l[0], l[1])).full() for l in list]
65 70
66 def store_items(self, node_id, items, publisher): 71 def store_items(self, node_id, items, publisher):
76 data = item.toXml() 81 data = item.toXml()
77 cursor.execute("""UPDATE items SET publisher=%s, data=%s 82 cursor.execute("""UPDATE items SET publisher=%s, data=%s
78 FROM nodes 83 FROM nodes
79 WHERE nodes.id = items.node_id AND 84 WHERE nodes.id = items.node_id AND
80 nodes.node = %s and items.item=%s""", 85 nodes.node = %s and items.item=%s""",
81 (publisher.encode('utf8'), 86 (publisher.full().encode('utf8'),
82 data.encode('utf8'), 87 data.encode('utf8'),
83 node_id.encode('utf8'), 88 node_id.encode('utf8'),
84 item["id"].encode('utf8'))) 89 item["id"].encode('utf8')))
85 if cursor.rowcount == 1: 90 if cursor.rowcount == 1:
86 return 91 return
87 92
88 cursor.execute("""INSERT INTO items (node_id, item, publisher, data) 93 cursor.execute("""INSERT INTO items (node_id, item, publisher, data)
89 SELECT id, %s, %s, %s FROM nodes WHERE node=%s""", 94 SELECT id, %s, %s, %s FROM nodes WHERE node=%s""",
90 (item["id"].encode('utf8'), 95 (item["id"].encode('utf8'),
91 publisher.encode('utf8'), 96 publisher.full().encode('utf8'),
92 data.encode('utf8'), 97 data.encode('utf8'),
93 node_id.encode('utf8'))) 98 node_id.encode('utf8')))
94 99
95 def add_subscription(self, node_id, subscriber, state): 100 def add_subscription(self, node_id, subscriber, state):
96 return self.dbpool.runInteraction(self._add_subscription, node_id, 101 return self.dbpool.runInteraction(self._add_subscription, node_id,
97 subscriber, state) 102 subscriber, state)
98 103
99 def _add_subscription(self, cursor, node_id, subscriber, state): 104 def _add_subscription(self, cursor, node_id, subscriber, state):
100 self._check_node_exists(cursor, node_id) 105 self._check_node_exists(cursor, node_id)
101 subscriber = jid.JID(subscriber)
102 userhost = subscriber.userhost() 106 userhost = subscriber.userhost()
103 resource = subscriber.resource or '' 107 resource = subscriber.resource or ''
104 108
105 try: 109 try:
106 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""", 110 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""",
129 userhost.encode('utf8'), 133 userhost.encode('utf8'),
130 resource.encode('utf8'))) 134 resource.encode('utf8')))
131 state = cursor.fetchone()[0] 135 state = cursor.fetchone()[0]
132 136
133 return {'node': node_id, 137 return {'node': node_id,
134 'jid': subscriber.full(), 138 'jid': subscriber,
135 'subscription': state} 139 'subscription': state}
136 140
137 def remove_subscription(self, node_id, subscriber): 141 def remove_subscription(self, node_id, subscriber):
138 return self.dbpool.runInteraction(self._remove_subscription, node_id, 142 return self.dbpool.runInteraction(self._remove_subscription, node_id,
139 subscriber) 143 subscriber)
140 144
141 def _remove_subscription(self, cursor, node_id, subscriber): 145 def _remove_subscription(self, cursor, node_id, subscriber):
142 self._check_node_exists(cursor, node_id) 146 self._check_node_exists(cursor, node_id)
143 subscriber = jid.JID(subscriber)
144 userhost = subscriber.userhost() 147 userhost = subscriber.userhost()
145 resource = subscriber.resource or '' 148 resource = subscriber.resource or ''
146 149
147 cursor.execute("""DELETE FROM subscriptions WHERE 150 cursor.execute("""DELETE FROM subscriptions WHERE
148 node_id=(SELECT id FROM nodes WHERE node=%s) AND 151 node_id=(SELECT id FROM nodes WHERE node=%s) AND
166 (node_id.encode('utf8'))) 169 (node_id.encode('utf8')))
167 except: 170 except:
168 raise backend.NodeExists 171 raise backend.NodeExists
169 172
170 cursor.execute("""SELECT 1 from entities where jid=%s""", 173 cursor.execute("""SELECT 1 from entities where jid=%s""",
171 (owner.encode('utf8'))) 174 (owner.full().encode('utf8')))
172 175
173 if not cursor.fetchone(): 176 if not cursor.fetchone():
174 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""", 177 cursor.execute("""INSERT INTO entities (jid) VALUES (%s)""",
175 (owner.encode('utf8'))) 178 (owner.full().encode('utf8')))
176 179
177 try: 180 try:
178 cursor.execute("""INSERT INTO affiliations 181 cursor.execute("""INSERT INTO affiliations
179 (node_id, entity_id, affiliation) 182 (node_id, entity_id, affiliation)
180 SELECT n.id, e.id, 'owner' FROM 183 SELECT n.id, e.id, 'owner' FROM
181 (SELECT id FROM nodes WHERE node=%s) AS n 184 (SELECT id FROM nodes WHERE node=%s) AS n
182 CROSS JOIN 185 CROSS JOIN
183 (SELECT id FROM entities WHERE jid=%s) AS e""", 186 (SELECT id FROM entities WHERE jid=%s) AS e""",
184 (node_id.encode('utf8'), 187 (node_id.encode('utf8'),
185 owner.encode('utf8'))) 188 owner.full().encode('utf8')))
186 except Exception, e: 189 except Exception, e:
187 print e 190 print e
188 191
189 return None 192 return None
190 193