Mercurial > libervia-pubsub

comparison idavoll/pgsql_storage.py @ 198:e404775b12df

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Change naming and spacing conventions to match Twisted's.
author Ralph Meijer <ralphm@ik.nu>
date Tue, 10 Jun 2008 11:31:49 +0000
parents 0d4474051eeb
children 77c61e2b8c75
comparison
equal deleted inserted replaced
197:9da5a95d408d 198:e404775b12df
21 port=port, 21 port=port,
22 cp_reconnect=True, 22 cp_reconnect=True,
23 client_encoding='utf-8' 23 client_encoding='utf-8'
24 ) 24 )
25 25
26 def get_node(self, node_id): 26
27 return self._dbpool.runInteraction(self._get_node, node_id) 27 def getNode(self, nodeIdentifier):
28 28 return self._dbpool.runInteraction(self._getNode, nodeIdentifier)
29 def _get_node(self, cursor, node_id): 29
30
31 def _getNode(self, cursor, nodeIdentifier):
30 configuration = {} 32 configuration = {}
31 cursor.execute("""SELECT persistent, deliver_payload, 33 cursor.execute("""SELECT persistent, deliver_payload,
32 send_last_published_item 34 send_last_published_item
33 FROM nodes 35 FROM nodes
34 WHERE node=%s""", 36 WHERE node=%s""",
35 (node_id,)) 37 (nodeIdentifier,))
36 try: 38 try:
37 (configuration["pubsub#persist_items"], 39 (configuration["pubsub#persist_items"],
38 configuration["pubsub#deliver_payloads"], 40 configuration["pubsub#deliver_payloads"],
39 configuration["pubsub#send_last_published_item"]) = \ 41 configuration["pubsub#send_last_published_item"]) = \
40 cursor.fetchone() 42 cursor.fetchone()
41 except TypeError: 43 except TypeError:
42 raise error.NodeNotFound() 44 raise error.NodeNotFound()
43 else: 45 else:
44 node = LeafNode(node_id, configuration) 46 node = LeafNode(nodeIdentifier, configuration)
45 node._dbpool = self._dbpool 47 node._dbpool = self._dbpool
46 return node 48 return node
47 49
48 def get_node_ids(self): 50
51 def getNodeIds(self):
49 d = self._dbpool.runQuery("""SELECT node from nodes""") 52 d = self._dbpool.runQuery("""SELECT node from nodes""")
50 d.addCallback(lambda results: [r[0] for r in results]) 53 d.addCallback(lambda results: [r[0] for r in results])
51 return d 54 return d
52 55
53 def create_node(self, node_id, owner, config=None): 56
54 return self._dbpool.runInteraction(self._create_node, node_id, owner) 57 def createNode(self, nodeIdentifier, owner, config=None):
55 58 return self._dbpool.runInteraction(self._createNode, nodeIdentifier,
56 def _create_node(self, cursor, node_id, owner): 59 owner)
57 node_id = node_id 60
61
62 def _createNode(self, cursor, nodeIdentifier, owner):
58 owner = owner.userhost() 63 owner = owner.userhost()
59 try: 64 try:
60 cursor.execute("""INSERT INTO nodes (node) VALUES (%s)""", 65 cursor.execute("""INSERT INTO nodes (node) VALUES (%s)""",
61 (node_id)) 66 (nodeIdentifier))
62 except cursor._pool.dbapi.OperationalError: 67 except cursor._pool.dbapi.OperationalError:
63 raise error.NodeExists() 68 raise error.NodeExists()
64 69
65 cursor.execute("""SELECT 1 from entities where jid=%s""", 70 cursor.execute("""SELECT 1 from entities where jid=%s""",
66 (owner)) 71 (owner))
73 (node_id, entity_id, affiliation) 78 (node_id, entity_id, affiliation)
74 SELECT n.id, e.id, 'owner' FROM 79 SELECT n.id, e.id, 'owner' FROM
75 (SELECT id FROM nodes WHERE node=%s) AS n 80 (SELECT id FROM nodes WHERE node=%s) AS n
76 CROSS JOIN 81 CROSS JOIN
77 (SELECT id FROM entities WHERE jid=%s) AS e""", 82 (SELECT id FROM entities WHERE jid=%s) AS e""",
78 (node_id, owner)) 83 (nodeIdentifier, owner))
79 84
80 def delete_node(self, node_id): 85
81 return self._dbpool.runInteraction(self._delete_node, node_id) 86 def deleteNode(self, nodeIdentifier):
82 87 return self._dbpool.runInteraction(self._deleteNode, nodeIdentifier)
83 def _delete_node(self, cursor, node_id): 88
89
90 def _deleteNode(self, cursor, nodeIdentifier):
84 cursor.execute("""DELETE FROM nodes WHERE node=%s""", 91 cursor.execute("""DELETE FROM nodes WHERE node=%s""",
85 (node_id,)) 92 (nodeIdentifier,))
86 93
87 if cursor.rowcount != 1: 94 if cursor.rowcount != 1:
88 raise error.NodeNotFound() 95 raise error.NodeNotFound()
89 96
90 def get_affiliations(self, entity): 97
98 def getAffiliations(self, entity):
91 d = self._dbpool.runQuery("""SELECT node, affiliation FROM entities 99 d = self._dbpool.runQuery("""SELECT node, affiliation FROM entities
92 JOIN affiliations ON 100 JOIN affiliations ON
93 (affiliations.entity_id=entities.id) 101 (affiliations.entity_id=entities.id)
94 JOIN nodes ON 102 JOIN nodes ON
95 (nodes.id=affiliations.node_id) 103 (nodes.id=affiliations.node_id)
96 WHERE jid=%s""", 104 WHERE jid=%s""",
97 (entity.userhost(),)) 105 (entity.userhost(),))
98 d.addCallback(lambda results: [tuple(r) for r in results]) 106 d.addCallback(lambda results: [tuple(r) for r in results])
99 return d 107 return d
100 108
101 def get_subscriptions(self, entity): 109
110 def getSubscriptions(self, entity):
102 d = self._dbpool.runQuery("""SELECT node, jid, resource, subscription 111 d = self._dbpool.runQuery("""SELECT node, jid, resource, subscription
103 FROM entities JOIN subscriptions ON 112 FROM entities JOIN subscriptions ON
104 (subscriptions.entity_id=entities.id) 113 (subscriptions.entity_id=entities.id)
105 JOIN nodes ON 114 JOIN nodes ON
106 (nodes.id=subscriptions.node_id) 115 (nodes.id=subscriptions.node_id)
107 WHERE jid=%s""", 116 WHERE jid=%s""",
108 (entity.userhost(),)) 117 (entity.userhost(),))
109 d.addCallback(self._convert_subscription_jids) 118 d.addCallback(self._convertSubscriptionJIDs)
110 return d 119 return d
111 120
112 def _convert_subscription_jids(self, subscriptions): 121
122 def _convertSubscriptionJIDs(self, subscriptions):
113 return [(node, 123 return [(node,
114 jid.internJID('%s/%s' % (subscriber, resource)), 124 jid.internJID('%s/%s' % (subscriber, resource)),
115 subscription) 125 subscription)
116 for node, subscriber, resource, subscription in subscriptions] 126 for node, subscriber, resource, subscription in subscriptions]
117 127
118 128
129
119 class Node: 130 class Node:
120 131
121 implements(iidavoll.INode) 132 implements(iidavoll.INode)
122 133
123 def __init__(self, node_id, config): 134 def __init__(self, nodeIdentifier, config):
124 self.id = node_id 135 self.nodeIdentifier = nodeIdentifier
125 self._config = config 136 self._config = config
126 137
127 def _check_node_exists(self, cursor): 138
139 def _checkNodeExists(self, cursor):
128 cursor.execute("""SELECT id FROM nodes WHERE node=%s""", 140 cursor.execute("""SELECT id FROM nodes WHERE node=%s""",
129 (self.id)) 141 (self.nodeIdentifier))
130 if not cursor.fetchone(): 142 if not cursor.fetchone():
131 raise error.NodeNotFound() 143 raise error.NodeNotFound()
132 144
133 def get_type(self): 145
134 return self.type 146 def getType(self):
135 147 return self.nodeType
136 def get_configuration(self): 148
149
150 def getConfiguration(self):
137 return self._config 151 return self._config
138 152
139 def set_configuration(self, options): 153
154 def setConfiguration(self, options):
140 config = copy.copy(self._config) 155 config = copy.copy(self._config)
141 156
142 for option in options: 157 for option in options:
143 if option in config: 158 if option in config:
144 config[option] = options[option] 159 config[option] = options[option]
145 160
146 d = self._dbpool.runInteraction(self._set_configuration, config) 161 d = self._dbpool.runInteraction(self._setConfiguration, config)
147 d.addCallback(self._set_cached_configuration, config) 162 d.addCallback(self._setCachedConfiguration, config)
148 return d 163 return d
149 164
150 def _set_configuration(self, cursor, config): 165
151 self._check_node_exists(cursor) 166 def _setConfiguration(self, cursor, config):
167 self._checkNodeExists(cursor)
152 cursor.execute("""UPDATE nodes SET persistent=%s, deliver_payload=%s, 168 cursor.execute("""UPDATE nodes SET persistent=%s, deliver_payload=%s,
153 send_last_published_item=%s 169 send_last_published_item=%s
154 WHERE node=%s""", 170 WHERE node=%s""",
155 (config["pubsub#persist_items"], 171 (config["pubsub#persist_items"],
156 config["pubsub#deliver_payloads"], 172 config["pubsub#deliver_payloads"],
157 config["pubsub#send_last_published_item"], 173 config["pubsub#send_last_published_item"],
158 self.id)) 174 self.nodeIdentifier))
159 175
160 def _set_cached_configuration(self, void, config): 176
177 def _setCachedConfiguration(self, void, config):
161 self._config = config 178 self._config = config
162 179
163 def get_meta_data(self): 180
181 def getMetaData(self):
164 config = copy.copy(self._config) 182 config = copy.copy(self._config)
165 config["pubsub#node_type"] = self.type 183 config["pubsub#node_type"] = self.nodeType
166 return config 184 return config
167 185
168 def get_affiliation(self, entity): 186
169 return self._dbpool.runInteraction(self._get_affiliation, entity) 187 def getAffiliation(self, entity):
170 188 return self._dbpool.runInteraction(self._getAffiliation, entity)
171 def _get_affiliation(self, cursor, entity): 189
172 self._check_node_exists(cursor) 190
191 def _getAffiliation(self, cursor, entity):
192 self._checkNodeExists(cursor)
173 cursor.execute("""SELECT affiliation FROM affiliations 193 cursor.execute("""SELECT affiliation FROM affiliations
174 JOIN nodes ON (node_id=nodes.id) 194 JOIN nodes ON (node_id=nodes.id)
175 JOIN entities ON (entity_id=entities.id) 195 JOIN entities ON (entity_id=entities.id)
176 WHERE node=%s AND jid=%s""", 196 WHERE node=%s AND jid=%s""",
177 (self.id, 197 (self.nodeIdentifier,
178 entity.userhost())) 198 entity.userhost()))
179 199
180 try: 200 try:
181 return cursor.fetchone()[0] 201 return cursor.fetchone()[0]
182 except TypeError: 202 except TypeError:
183 return None 203 return None
184 204
185 def get_subscription(self, subscriber): 205
186 return self._dbpool.runInteraction(self._get_subscription, subscriber) 206 def getSubscription(self, subscriber):
187 207 return self._dbpool.runInteraction(self._getSubscription, subscriber)
188 def _get_subscription(self, cursor, subscriber): 208
189 self._check_node_exists(cursor) 209
210 def _getSubscription(self, cursor, subscriber):
211 self._checkNodeExists(cursor)
190 212
191 userhost = subscriber.userhost() 213 userhost = subscriber.userhost()
192 resource = subscriber.resource or '' 214 resource = subscriber.resource or ''
193 215
194 cursor.execute("""SELECT subscription FROM subscriptions 216 cursor.execute("""SELECT subscription FROM subscriptions
195 JOIN nodes ON (nodes.id=subscriptions.node_id) 217 JOIN nodes ON (nodes.id=subscriptions.node_id)
196 JOIN entities ON 218 JOIN entities ON
197 (entities.id=subscriptions.entity_id) 219 (entities.id=subscriptions.entity_id)
198 WHERE node=%s AND jid=%s AND resource=%s""", 220 WHERE node=%s AND jid=%s AND resource=%s""",
199 (self.id, 221 (self.nodeIdentifier,
200 userhost, 222 userhost,
201 resource)) 223 resource))
202 try: 224 try:
203 return cursor.fetchone()[0] 225 return cursor.fetchone()[0]
204 except TypeError: 226 except TypeError:
205 return None 227 return None
206 228
207 def add_subscription(self, subscriber, state): 229
208 return self._dbpool.runInteraction(self._add_subscription, subscriber, 230 def addSubscription(self, subscriber, state):
231 return self._dbpool.runInteraction(self._addSubscription, subscriber,
209 state) 232 state)
210 233
211 def _add_subscription(self, cursor, subscriber, state): 234
212 self._check_node_exists(cursor) 235 def _addSubscription(self, cursor, subscriber, state):
236 self._checkNodeExists(cursor)
213 237
214 userhost = subscriber.userhost() 238 userhost = subscriber.userhost()
215 resource = subscriber.resource or '' 239 resource = subscriber.resource or ''
216 240
217 try: 241 try:
227 (SELECT id FROM nodes WHERE node=%s) AS n 251 (SELECT id FROM nodes WHERE node=%s) AS n
228 CROSS JOIN 252 CROSS JOIN
229 (SELECT id FROM entities WHERE jid=%s) AS e""", 253 (SELECT id FROM entities WHERE jid=%s) AS e""",
230 (resource, 254 (resource,
231 state, 255 state,
232 self.id, 256 self.nodeIdentifier,
233 userhost)) 257 userhost))
234 except cursor._pool.dbapi.OperationalError: 258 except cursor._pool.dbapi.OperationalError:
235 raise error.SubscriptionExists() 259 raise error.SubscriptionExists()
236 260
237 def remove_subscription(self, subscriber): 261
238 return self._dbpool.runInteraction(self._remove_subscription, 262 def removeSubscription(self, subscriber):
263 return self._dbpool.runInteraction(self._removeSubscription,
239 subscriber) 264 subscriber)
240 265
241 def _remove_subscription(self, cursor, subscriber): 266
242 self._check_node_exists(cursor) 267 def _removeSubscription(self, cursor, subscriber):
268 self._checkNodeExists(cursor)
243 269
244 userhost = subscriber.userhost() 270 userhost = subscriber.userhost()
245 resource = subscriber.resource or '' 271 resource = subscriber.resource or ''
246 272
247 cursor.execute("""DELETE FROM subscriptions WHERE 273 cursor.execute("""DELETE FROM subscriptions WHERE
248 node_id=(SELECT id FROM nodes WHERE node=%s) AND 274 node_id=(SELECT id FROM nodes WHERE node=%s) AND
249 entity_id=(SELECT id FROM entities WHERE jid=%s) 275 entity_id=(SELECT id FROM entities WHERE jid=%s)
250 AND resource=%s""", 276 AND resource=%s""",
251 (self.id, 277 (self.nodeIdentifier,
252 userhost, 278 userhost,
253 resource)) 279 resource))
254 if cursor.rowcount != 1: 280 if cursor.rowcount != 1:
255 raise error.NotSubscribed() 281 raise error.NotSubscribed()
256 282
257 return None 283 return None
258 284
259 def get_subscribers(self): 285
260 d = self._dbpool.runInteraction(self._get_subscribers) 286 def getSubscribers(self):
261 d.addCallback(self._convert_to_jids) 287 d = self._dbpool.runInteraction(self._getSubscribers)
288 d.addCallback(self._convertToJIDs)
262 return d 289 return d
263 290
264 def _get_subscribers(self, cursor): 291
265 self._check_node_exists(cursor) 292 def _getSubscribers(self, cursor):
293 self._checkNodeExists(cursor)
266 cursor.execute("""SELECT jid, resource FROM subscriptions 294 cursor.execute("""SELECT jid, resource FROM subscriptions
267 JOIN nodes ON (node_id=nodes.id) 295 JOIN nodes ON (node_id=nodes.id)
268 JOIN entities ON (entity_id=entities.id) 296 JOIN entities ON (entity_id=entities.id)
269 WHERE node=%s AND 297 WHERE node=%s AND
270 subscription='subscribed'""", 298 subscription='subscribed'""",
271 (self.id,)) 299 (self.nodeIdentifier,))
272 return cursor.fetchall() 300 return cursor.fetchall()
273 301
274 def _convert_to_jids(self, list): 302
303 def _convertToJIDs(self, list):
275 return [jid.internJID("%s/%s" % (l[0], l[1])) for l in list] 304 return [jid.internJID("%s/%s" % (l[0], l[1])) for l in list]
276 305
277 def is_subscribed(self, entity): 306
278 return self._dbpool.runInteraction(self._is_subscribed, entity) 307 def isSubscribed(self, entity):
279 308 return self._dbpool.runInteraction(self._isSubscribed, entity)
280 def _is_subscribed(self, cursor, entity): 309
281 self._check_node_exists(cursor) 310
311 def _isSubscribed(self, cursor, entity):
312 self._checkNodeExists(cursor)
282 313
283 cursor.execute("""SELECT 1 FROM entities 314 cursor.execute("""SELECT 1 FROM entities
284 JOIN subscriptions ON 315 JOIN subscriptions ON
285 (entities.id=subscriptions.entity_id) 316 (entities.id=subscriptions.entity_id)
286 JOIN nodes ON 317 JOIN nodes ON
287 (nodes.id=subscriptions.node_id) 318 (nodes.id=subscriptions.node_id)
288 WHERE entities.jid=%s 319 WHERE entities.jid=%s
289 AND node=%s AND subscription='subscribed'""", 320 AND node=%s AND subscription='subscribed'""",
290 (entity.userhost(), 321 (entity.userhost(),
291 self.id)) 322 self.nodeIdentifier))
292 323
293 return cursor.fetchone() is not None 324 return cursor.fetchone() is not None
294 325
295 def get_affiliations(self): 326
296 return self._dbpool.runInteraction(self._get_affiliations) 327 def getAffiliations(self):
297 328 return self._dbpool.runInteraction(self._getAffiliations)
298 def _get_affiliations(self, cursor): 329
299 self._check_node_exists(cursor) 330
331 def _getAffiliations(self, cursor):
332 self._checkNodeExists(cursor)
300 333
301 cursor.execute("""SELECT jid, affiliation FROM nodes 334 cursor.execute("""SELECT jid, affiliation FROM nodes
302 JOIN affiliations ON 335 JOIN affiliations ON
303 (nodes.id = affiliations.node_id) 336 (nodes.id = affiliations.node_id)
304 JOIN entities ON 337 JOIN entities ON
305 (affiliations.entity_id = entities.id) 338 (affiliations.entity_id = entities.id)
306 WHERE node=%s""", 339 WHERE node=%s""",
307 self.id) 340 self.nodeIdentifier)
308 result = cursor.fetchall() 341 result = cursor.fetchall()
309 342
310 return [(jid.internJID(r[0]), r[1]) for r in result] 343 return [(jid.internJID(r[0]), r[1]) for r in result]
311 344
345
346
312 class LeafNodeMixin: 347 class LeafNodeMixin:
313 348
314 type = 'leaf' 349 nodeType = 'leaf'
315 350
316 def store_items(self, items, publisher): 351 def storeItems(self, items, publisher):
317 return self._dbpool.runInteraction(self._store_items, items, publisher) 352 return self._dbpool.runInteraction(self._storeItems, items, publisher)
318 353
319 def _store_items(self, cursor, items, publisher): 354
320 self._check_node_exists(cursor) 355 def _storeItems(self, cursor, items, publisher):
356 self._checkNodeExists(cursor)
321 for item in items: 357 for item in items:
322 self._store_item(cursor, item, publisher) 358 self._storeItem(cursor, item, publisher)
323 359
324 def _store_item(self, cursor, item, publisher): 360
361 def _storeItem(self, cursor, item, publisher):
325 data = item.toXml() 362 data = item.toXml()
326 cursor.execute("""UPDATE items SET date=now(), publisher=%s, data=%s 363 cursor.execute("""UPDATE items SET date=now(), publisher=%s, data=%s
327 FROM nodes 364 FROM nodes
328 WHERE nodes.id = items.node_id AND 365 WHERE nodes.id = items.node_id AND
329 nodes.node = %s and items.item=%s""", 366 nodes.node = %s and items.item=%s""",
330 (publisher.full(), 367 (publisher.full(),
331 data, 368 data,
332 self.id, 369 self.nodeIdentifier,
333 item["id"])) 370 item["id"]))
334 if cursor.rowcount == 1: 371 if cursor.rowcount == 1:
335 return 372 return
336 373
337 cursor.execute("""INSERT INTO items (node_id, item, publisher, data) 374 cursor.execute("""INSERT INTO items (node_id, item, publisher, data)
338 SELECT id, %s, %s, %s FROM nodes WHERE node=%s""", 375 SELECT id, %s, %s, %s FROM nodes WHERE node=%s""",
339 (item["id"], 376 (item["id"],
340 publisher.full(), 377 publisher.full(),
341 data, 378 data,
342 self.id)) 379 self.nodeIdentifier))
343 380
344 def remove_items(self, item_ids): 381
345 return self._dbpool.runInteraction(self._remove_items, item_ids) 382 def removeItems(self, itemIdentifiers):
346 383 return self._dbpool.runInteraction(self._removeItems, itemIdentifiers)
347 def _remove_items(self, cursor, item_ids): 384
348 self._check_node_exists(cursor) 385
386 def _removeItems(self, cursor, itemIdentifiers):
387 self._checkNodeExists(cursor)
349 388
350 deleted = [] 389 deleted = []
351 390
352 for item_id in item_ids: 391 for itemIdentifier in itemIdentifiers:
353 cursor.execute("""DELETE FROM items WHERE 392 cursor.execute("""DELETE FROM items WHERE
354 node_id=(SELECT id FROM nodes WHERE node=%s) AND 393 node_id=(SELECT id FROM nodes WHERE node=%s) AND
355 item=%s""", 394 item=%s""",
356 (self.id, 395 (self.nodeIdentifier,
357 item_id)) 396 itemIdentifier))
358 397
359 if cursor.rowcount: 398 if cursor.rowcount:
360 deleted.append(item_id) 399 deleted.append(itemIdentifier)
361 400
362 return deleted 401 return deleted
363 402
364 def get_items(self, max_items=None): 403
365 return self._dbpool.runInteraction(self._get_items, max_items) 404 def getItems(self, maxItems=None):
366 405 return self._dbpool.runInteraction(self._getItems, maxItems)
367 def _get_items(self, cursor, max_items): 406
368 self._check_node_exists(cursor) 407
408 def _getItems(self, cursor, maxItems):
409 self._checkNodeExists(cursor)
369 query = """SELECT data FROM nodes JOIN items ON 410 query = """SELECT data FROM nodes JOIN items ON
370 (nodes.id=items.node_id) 411 (nodes.id=items.node_id)
371 WHERE node=%s ORDER BY date DESC""" 412 WHERE node=%s ORDER BY date DESC"""
372 if max_items: 413 if maxItems:
373 cursor.execute(query + " LIMIT %s", 414 cursor.execute(query + " LIMIT %s",
374 (self.id, 415 (self.nodeIdentifier,
375 max_items)) 416 maxItems))
376 else: 417 else:
377 cursor.execute(query, (self.id)) 418 cursor.execute(query, (self.nodeIdentifier))
378 419
379 result = cursor.fetchall() 420 result = cursor.fetchall()
380 return [unicode(r[0], 'utf-8') for r in result] 421 return [unicode(r[0], 'utf-8') for r in result]
381 422
382 def get_items_by_id(self, item_ids): 423
383 return self._dbpool.runInteraction(self._get_items_by_id, item_ids) 424 def getItemsById(self, itemIdentifiers):
384 425 return self._dbpool.runInteraction(self._getItemsById, itemIdentifiers)
385 def _get_items_by_id(self, cursor, item_ids): 426
386 self._check_node_exists(cursor) 427
428 def _getItemsById(self, cursor, itemIdentifiers):
429 self._checkNodeExists(cursor)
387 items = [] 430 items = []
388 for item_id in item_ids: 431 for itemIdentifier in itemIdentifiers:
389 cursor.execute("""SELECT data FROM nodes JOIN items ON 432 cursor.execute("""SELECT data FROM nodes JOIN items ON
390 (nodes.id=items.node_id) 433 (nodes.id=items.node_id)
391 WHERE node=%s AND item=%s""", 434 WHERE node=%s AND item=%s""",
392 (self.id, 435 (self.nodeIdentifier,
393 item_id)) 436 itemIdentifier))
394 result = cursor.fetchone() 437 result = cursor.fetchone()
395 if result: 438 if result:
396 items.append(unicode(result[0], 'utf-8')) 439 items.append(unicode(result[0], 'utf-8'))
397 return items 440 return items
398 441
442
399 def purge(self): 443 def purge(self):
400 return self._dbpool.runInteraction(self._purge) 444 return self._dbpool.runInteraction(self._purge)
401 445
446
402 def _purge(self, cursor): 447 def _purge(self, cursor):
403 self._check_node_exists(cursor) 448 self._checkNodeExists(cursor)
404 449
405 cursor.execute("""DELETE FROM items WHERE 450 cursor.execute("""DELETE FROM items WHERE
406 node_id=(SELECT id FROM nodes WHERE node=%s)""", 451 node_id=(SELECT id FROM nodes WHERE node=%s)""",
407 (self.id,)) 452 (self.nodeIdentifier,))
453
454
408 455
409 class LeafNode(Node, LeafNodeMixin): 456 class LeafNode(Node, LeafNodeMixin):
410 457
411 implements(iidavoll.ILeafNode) 458 implements(iidavoll.ILeafNode)