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

title: str

properties: dict[str, str]

class search.indexer.Indexer(mappings: TypeMappingRegistry, languages: set[str] | None = None)

mappings

languages

index(tasks: list[IndexTask] | IndexTask, session: sqlalchemy.orm.Session) → 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.

The upsert is executed directly on the given session’s transaction, without a savepoint of its own, so the caller is responsible for committing or rolling back.

Parameters:

• tasks – A list of tasks to index

• session – Supply an active session

Returns:

True if the indexing was successful, False otherwise

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

process(tasks: collections.abc.Iterable[Task], session: sqlalchemy.orm.Session) → 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(session: sqlalchemy.orm.Session) → None

Immediately delete all records in search index table.

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

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

name

mapping

title_property = None

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], title_property: str | None = None, 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(indexer: Indexer, 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.

indexer

mappings

detector

max_queue_size = 0

stopped = False

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

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

on_delete(schema: str, session: sqlalchemy.orm.Session, obj: object) → None

on_transaction_join(schema: str, session: sqlalchemy.orm.Session) → None

put(session: sqlalchemy.orm.Session, translation: Task) → None

index_task(schema: str, obj: onegov.search.Searchable) → IndexTask | None

delete_task(schema: str, obj: onegov.search.Searchable) → DeleteTask

index(schema: str, obj: onegov.search.Searchable, session: sqlalchemy.orm.Session | None = None) → None

Creates or updates index for the given object

delete(schema: str, obj: onegov.search.Searchable, session: sqlalchemy.orm.Session | None = None) → None

Deletes index of the given object