libervia-pubsub: sat_pubsub/pgsql

comparison sat_pubsub/pgsql_storage.py @ 430:5a0ada3b61ca

Full-Text Search implementation: /!\ pgsql schema needs to be updated /!\ /!\ Minimal PostgreSQL required version is now 12 /!\ A new options is available to specify main language of a node. By default a `generic` language is used (which uses the `simple` configuration in PostgreSQL). When a node owner changes the language, the index is rebuilt accordingly. It is possible to have item specific language for multilingual nodes (but for the moment the search is done with node language, so the results won't be good). If an item language is explicitely set in `item_languages`, the FTS configuration won't be affected by node FTS language setting. Search is parsed with `websearch_to_tsquery` for now, but this parser doesn't handle prefix matching, so it may be replaced in the future. SetConfiguration now only updates the modified values, this avoid triggering the FTS re-indexing on each config change. `_checkNodeExists` is not called anymore as we can check if a row has been modified to see if the node exists, this avoid a useless query. Item storing has been slighly improved with a useless SELECT and condition removed. To avoid 2 schema updates in a row, the `sat_pubsub_update_5_6.sql` file also prepares the implementation of XEP-0346 by updating nodes with a schema and creating the suitable template nodes.

author	Goffi <goffi@goffi.org>
date	Fri, 11 Dec 2020 17:18:52 +0100
parents	0526073ff2ab
children	920440200570

comparison

equal deleted inserted replaced

-:0526073ff2ab
+:5a0ada3b61ca
 # parseXml manage str, but we get unicode
 parseXml = lambda unicode_data: generic.parseXml(unicode_data.encode('utf-8'))
 ITEMS_SEQ_NAME = 'node_{node_id}_seq'
 PEP_COL_NAME = 'pep'
-CURRENT_VERSION = '5'
+CURRENT_VERSION = '6'
 # retrieve the maximum integer item id + 1
 NEXT_ITEM_ID_QUERY = r"SELECT COALESCE(max(item::integer)+1,1) as val from items where node_id={node_id} and item ~ E'^\\d+$'"
 def withPEP(query, values, pep, recipient):
 @implementer(iidavoll.IStorage)
 class Storage:
+fts_languages = ['generic']
 defaultConfig = {
 'leaf': {
 const.OPT_PERSIST_ITEMS: True,
 const.OPT_MAX_ITEMS: 'max',
 const.OPT_DELIVER_PAYLOADS: True,
 const.OPT_SEND_LAST_PUBLISHED_ITEM: 'on_sub',
 const.OPT_ACCESS_MODEL: const.VAL_AMODEL_DEFAULT,
 const.OPT_PUBLISH_MODEL: const.VAL_PMODEL_DEFAULT,
 const.OPT_SERIAL_IDS: False,
 const.OPT_CONSISTENT_PUBLISHER: False,
+const.OPT_FTS_LANGUAGE: const.VAL_FTS_GENERIC,
 },
 'collection': {
 const.OPT_DELIVER_PAYLOADS: True,
 const.OPT_SEND_LAST_PUBLISHED_ITEM: 'on_sub',
 const.OPT_ACCESS_MODEL: const.VAL_AMODEL_DEFAULT,
 const.OPT_SEND_LAST_PUBLISHED_ITEM: row[6],
 const.OPT_ACCESS_MODEL:row[7],
 const.OPT_PUBLISH_MODEL:row[8],
 const.OPT_SERIAL_IDS:row[9],
 const.OPT_CONSISTENT_PUBLISHER:row[10],
+const.OPT_FTS_LANGUAGE: row[11],
 }
-schema = row[11]
+schema = row[12]
 if schema is not None:
 schema = parseXml(schema)
 node = LeafNode(row[0], row[1], configuration, schema)
 node.dbpool = self.dbpool
 return node
 node.dbpool = self.dbpool
 return node
 else:
 raise BadRequest(text="Unknown node type !")
+def getFTSLanguages(self):
+"""Get list of available languages for full text search"""
+return self.dbpool.runInteraction(self._getFTSLanguages)
+def _getFTSLanguages(self, cursor):
+cursor.execute("SELECT cfgname FROM pg_ts_config")
+result = [r.cfgname for r in cursor.fetchall()]
+result.remove("simple")
+result.insert(0, "generic")
+Node.fts_languages = self.fts_languages = result
+return result
 def getNodeById(self, nodeDbId):
 """Get node using database ID insted of pubsub identifier
 @param nodeDbId(unicode): database ID
 """
 send_last_published_item,
 access_model,
 publish_model,
 serial_ids,
 consistent_publisher,
+fts_language,
 schema::text,
 pep
 FROM nodes
 WHERE node_id=%s""",
 (nodeDbId,))
 send_last_published_item,
 access_model,
 publish_model,
 serial_ids,
 consistent_publisher,
+fts_language,
 schema::text,
 pep
 FROM nodes
 WHERE node=%s""",
 (nodeIdentifier,), pep, recipient))
 send_last_published_item,
 access_model,
 publish_model,
 serial_ids,
 consistent_publisher,
+fts_language,
 schema,
 pep)
 VALUES
-(%s, 'leaf', %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
+(%s, 'leaf', %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
 (nodeIdentifier,
 config['pubsub#persist_items'],
 config['pubsub#deliver_payloads'],
 config['pubsub#send_last_published_item'],
 config[const.OPT_ACCESS_MODEL],
 config[const.OPT_PUBLISH_MODEL],
 config[const.OPT_SERIAL_IDS],
 config[const.OPT_CONSISTENT_PUBLISHER],
+config[const.OPT_FTS_LANGUAGE],
 schema,
 recipient.userhost() if pep else None
 )
 )
 except cursor._pool.dbapi.IntegrityError as e:
 @param cursor(): current db cursor
 @param node_id(unicode): database ID of the node
 @param old_config(dict): config of the node before the change
 @param new_config(dict): new options that will be changed
 """
+if const.OPT_SERIAL_IDS not in new_config:
+return
 serial_ids = new_config[const.OPT_SERIAL_IDS]
 if serial_ids != old_config[const.OPT_SERIAL_IDS]:
 # serial_ids option has been modified,
 # we need to handle corresponding sequence
 next_val = next_val))
 else:
 cursor.execute("DROP SEQUENCE IF EXISTS {seq_name}".format(seq_name = seq_name))
 def setConfiguration(self, options):
-config = copy.copy(self._config)
+to_delete = []
+for option, value in options.items():
-for option in options:
+try:
-if option in config:
+if self._config[option] == value:
-config[option] = options[option]
+to_delete.append(option)
+except KeyError:
-if config[const.OPT_MAX_ITEMS] == "max":
+raise BadRequest(text=f"Invalid option: {option!r}")
+for option in to_delete:
+options.remove(option)
+if not options:
+return
+if options.get(const.OPT_MAX_ITEMS) == "max":
 # XXX: "max" is default value for config we must convert
 #   it to an interger. See backend's _doSetNodeConfiguration comment
-config[const.OPT_MAX_ITEMS] = "0"
+options[const.OPT_MAX_ITEMS] = "0"
-d = self.dbpool.runInteraction(self._setConfiguration, config)
+if ((const.OPT_FTS_LANGUAGE in options
-d.addCallback(self._setCachedConfiguration, config)
+and options[const.OPT_FTS_LANGUAGE] not in self.fts_languages)):
+raise BadRequest(text=
+f"invalid {const.OPT_FTS_LANGUAGE} value: "
+f"{options[const.OPT_FTS_LANGUAGE]!r}"
+)
+d = self.dbpool.runInteraction(self._setConfiguration, options)
+d.addCallback(self._updateCachedConfiguration, options)
 return d
-def _setConfiguration(self, cursor, config):
+def _setConfiguration(self, cursor, options):
-self._checkNodeExists(cursor)
+self._configurationTriggers(cursor, self.nodeDbId, self._config, options)
-self._configurationTriggers(cursor, self.nodeDbId, self._config, config)
+# options names all follow the scheme "pubsub#{col_name}"
-cursor.execute("""UPDATE nodes SET persist_items=%s,
+col_names = (o[7:] for o in options)
-max_items=%s,
+values = ','.join(f"{name}=%s" for name in col_names)
-deliver_payloads=%s,
+cursor.execute(f"UPDATE nodes SET {values} WHERE node_id=%s",
-send_last_published_item=%s,
+(*options.values(), self.nodeDbId))
-access_model=%s,
+if cursor.rowcount == 0:
-publish_model=%s,
+raise error.NodeNotFound()
-serial_ids=%s,
-consistent_publisher=%s
+def _updateCachedConfiguration(self, __, options):
-WHERE node_id=%s""",
+self._config.update(options)
-(config[const.OPT_PERSIST_ITEMS],
-config[const.OPT_MAX_ITEMS],
-config[const.OPT_DELIVER_PAYLOADS],
-config[const.OPT_SEND_LAST_PUBLISHED_ITEM],
-config[const.OPT_ACCESS_MODEL],
-config[const.OPT_PUBLISH_MODEL],
-config[const.OPT_SERIAL_IDS],
-config[const.OPT_CONSISTENT_PUBLISHER],
-self.nodeDbId))
-def _setCachedConfiguration(self, void, config):
-self._config = config
 def getSchema(self):
 return self._schema
 def setSchema(self, schema):
 cursor.execute("""UPDATE nodes SET schema=%s
 WHERE node_id=%s""",
 (schema.toXml() if schema else None,
 self.nodeDbId))
-def _setCachedSchema(self, void, schema):
+def _setCachedSchema(self, __, schema):
 self._schema = schema
 def getMetaData(self):
 config = copy.copy(self._config)
 config["pubsub#node_type"] = self.nodeType
 #   this case should be rare enough to consider this situation acceptable)
 # - if item insertion fail and the item is not new, we do an update
 # - in other cases, exception is raised
 item, access_model, item_config = item_data.item, item_data.access_model, item_data.config
 data = item.toXml()
+data_fts_cfg = self._config[const.OPT_FTS_LANGUAGE]
-insert_query = """INSERT INTO items (node_id, item, publisher, data, access_model)
+if data_fts_cfg == const.VAL_FTS_GENERIC:
-SELECT %s, %s, %s, %s, %s FROM nodes
+data_fts_cfg = "simple"
-WHERE node_id=%s
-RETURNING item_id"""
+insert_query = (
+"INSERT INTO items(node_id, item, publisher, data, access_model, "
+"data_fts_cfg) VALUES (%s, %s, %s, %s, %s, %s) RETURNING item_id"
+)
 insert_data = [self.nodeDbId,
 item["id"],
 publisher.full(),
 data,
 access_model,
-self.nodeDbId]
+data_fts_cfg,
+]
 try:
 cursor.execute(insert_query, insert_data)
 except cursor._pool.dbapi.IntegrityError as e:
 if e.pgcode != "23505":
 # but if we have not serial_ids, we have a real problem
 raise e
 else:
 # this is an update
 cursor.execute("""UPDATE items SET updated=now(), publisher=%s, data=%s
-FROM nodes
+WHERE node_id=%s AND items.item=%s
-WHERE nodes.node_id = items.node_id AND
-nodes.node_id = %s and items.item=%s
 RETURNING item_id""",
 (publisher.full(),
 data,
 self.nodeDbId,
 item["id"]))
 query_filters.append("AND publisher=%s")
 args.append(filter_.value)
 else:
 query_filters.append("AND publisher LIKE %s")
 args.append("{}%".format(filter_.value))
+elif filter_.var == const.MAM_FILTER_FTS:
+fts_cfg = self._config[const.OPT_FTS_LANGUAGE]
+if fts_cfg == const.VAL_FTS_GENERIC:
+fts_cfg = "simple"
+query_filters.append(
+"AND data_fts @@ websearch_to_tsquery(%s, %s)"
+)
+args.append(fts_cfg)
+args.append(filter_.value)
 elif filter_.var == const.MAM_FILTER_CATEGORY:
 query.append("LEFT JOIN item_categories USING (item_id)")
 query_filters.append("AND category=%s")
 args.append(filter_.value)
 else:

Mercurial > libervia-pubsub

comparison sat_pubsub/pgsql_storage.py @ 430:5a0ada3b61ca