search.indexer

Classes

IndexTask	dict() -> new empty dictionary
Indexer
TypeMapping
TypeMappingRegistry
ORMLanguageDetector	Detects languages with the help of lingua-language-detector.
ORMEventTranslator	Handles the onegov.core orm events, translates them into indexing

Module Contents

class search.indexer.IndexTask

Bases: TypedDict

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s

(key, value) pairs

dict(iterable) -> new dictionary initialized as if via:

d = {} for k, v in iterable:

d[k] = v

dict(**kwargs) -> new dictionary initialized with the name=value pairs

in the keyword argument list. For example: dict(one=1, two=2)

action: Literal['index']

id: uuid.UUID | str | int

id_key: str

schema: str

tablename: str

owner_type: str

language: str

access: str

public: bool

suggestion: list[str]

tags: list[str]

last_change: datetime.datetime | None

publication_start: datetime.datetime | None

publication_end: datetime.datetime | None

properties: dict[str, str]

class search.indexer.Indexer(mappings: TypeMappingRegistry, queue: queue.Queue[Task], engine: sqlalchemy.engine.Engine, languages: set[str] | None = None)

queue: queue.Queue[Task]

mappings

engine

languages

index(tasks: list[IndexTask] | IndexTask, session: sqlalchemy.orm.Session | None = None) → bool

Update the ‘search_index’ table (full text search index) of the given object(s)/task(s).

In case of a bunch of tasks we are assuming they are all from the same schema and table in order to optimize the indexing process.

When a session is passed we use that session’s transaction context and use a savepoint instead of our own transaction to perform the action.

Parameters:

• tasks – A list of tasks to index

• session – Supply an active session

Returns:

True if the indexing was successful, False otherwise

execute_statement(session: sqlalchemy.orm.Session | None, schema: str, stmt: sqlalchemy.sql.expression.Executable, params: list[dict[str, Any]] | None = None) → None

delete(tasks: list[IndexTask] | IndexTask, session: sqlalchemy.orm.Session | None = None) → bool

process(session: sqlalchemy.orm.Session | None = None) → int

Processes the queue in bulk.

Gathers all tasks and groups them by action and owner type.

Returns the number of successfully processed batches.

delete_search_index(schema: str) → None

Delete all records in search index table of the given schema.

class search.indexer.TypeMapping(name: str, mapping: dict[str, Any], model: type[onegov.search.Searchable] | None = None)

__slots__ = ('name', 'mapping', 'model')

name

mapping

model = None

class search.indexer.TypeMappingRegistry

mappings: dict[str, TypeMapping]

__getitem__(key: str) → TypeMapping

__iter__() → collections.abc.Iterator[TypeMapping]

register_orm_base(base: type[object]) → None

Takes the given SQLAlchemy base and registers all Searchable objects.

register_type(type_name: str, mapping: dict[str, Any], model: type[onegov.search.Searchable] | None = None) → None

Registers the given type with the given mapping. The mapping is as dictionary representing the part below the mappings/type_name.

See:

https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html#mappings

When the mapping changes, a new index is created internally and the alias to this index (the external name of the index) is pointed to this new index.

As a consequence, a change in the mapping requires a reindex.

property registered_fields: set[str]

Goes through all the registered types and returns the a set with all fields used by the mappings.

class search.indexer.ORMLanguageDetector(supported_languages: collections.abc.Sequence[str])

Bases: onegov.search.utils.LanguageDetector

Detects languages with the help of lingua-language-detector.

html_strip_expression

localized_properties(obj: onegov.search.Searchable) → collections.abc.Iterator[str]

localized_texts(obj: onegov.search.Searchable, max_chars: int | None = None) → collections.abc.Iterator[str]

detect_object_language(obj: onegov.search.Searchable) → str

class search.indexer.ORMEventTranslator(mappings: TypeMappingRegistry, max_queue_size: int = 0, languages: collections.abc.Sequence[str] = ('de', 'fr', 'en'))

Handles the onegov.core orm events, translates them into indexing actions and puts the result into a queue for the indexer to consume.

The queue may be limited. Once the limit is reached, new events are no longer processed and an error is logged.

queue: queue.Queue[Task]

mappings

detector

stopped = False

on_insert(schema: str, obj: object) → None

on_update(schema: str, obj: object) → None

on_delete(schema: str, obj: object) → None

put(translation: Task) → None

index(schema: str, obj: onegov.search.Searchable) → None

Creates or updates index for the given object

delete(schema: str, obj: onegov.search.Searchable) → None

Deletes index of the given object