Indexing API

index_bulk transforms entities with format_entity and bulk-indexes them with an Indexer. The indexer is tuned by the indexing settings.

from followthemoney import model
from openaleph_search.index.entities import index_bulk

entity = model.make_entity("Person")
entity.id = "jane-doe"
entity.add("name", "Jane Doe")
stats = index_bulk("my_dataset", [entity], sync=True)

With index_namespace_ids (the default) the indexed ID is signed with the dataset, so look entities up by the IDs a search returns.

Write

`openaleph_search.index.entities.index_bulk(dataset, entities, sync=False, **kwargs)`

Transform and index entities.

Parameters:

Name	Type	Description	Default
`dataset`	`str`	Dataset of the entities.	required
`entities`	`Iterable[EntityLike]`	Proxies, `EntityPayload`s or trusted entity dicts, see `format_entity`.	required
`sync`		Refresh, so the entities are searchable on return.	`False`
`**kwargs`		Passed to `Indexer`.	`{}`

Returns:

Type	Description
`IndexStats`	Counts of the run.

`openaleph_search.index.entities.index_proxy(dataset, proxy, sync=False, **kwargs)`

Index one entity and delete it from the indexes of other schemata, with the short max_retries budget of a request.

Parameters:

Name	Type	Description	Default
`dataset`	`str`	Dataset of the entity.	required
`proxy`	`EntityProxy`	The entity.	required
`sync`		Refresh, so the entity is searchable on return.	`False`
`**kwargs`		Passed to `Indexer`.	`{}`

Returns:

Type	Description
`IndexStats`	Counts of the run.

`openaleph_search.index.entities.delete_entity(entity_id, exclude=None, sync=False)`

Delete an entity from the index.

Parameters:

Name	Description	Default
`entity_id`	The entity ID.	required
`exclude`	Keep it in the write index of this schema.	`None`
`sync`	Refresh, so the deletion is visible on return.	`False`

`openaleph_search.index.indexer.Indexer(dataset=None, chunk_size=None, batch_bytes=None, concurrency=None, sync=False, max_retries=None, **context)`

Transform entities and bulk-index them from a single process: the transform runs inline while the bulk requests are in flight.

Parameters:

Name	Type	Description	Default
`dataset`	`str \| None`	Dataset of the entities, required by `index`.	`None`
`chunk_size`	`int \| None`	Entities per batch, defaults to `indexer_chunk_size`.	`None`
`batch_bytes`	`int \| None`	Payload bytes per batch, defaults to `indexer_batch_bytes`.	`None`
`concurrency`	`int \| None`	Upper bound on bulk requests in flight, defaults to `indexer_concurrency`.	`None`
`sync`	`bool \| None`	Refresh, so the entities are searchable on return.	`False`
`max_retries`	`int \| None`	Retries of a rejected batch, defaults to `indexer_max_retries`.	`None`
`**context`	`Any`	Passed to `format_entity`.	`{}`

`index(entities)`

Transform and index a stream of entities.

Parameters:

Name	Type	Description	Default
`entities`	`Iterable[EntityLike]`	Proxies, `EntityPayload`s or trusted entity dicts, see `format_entity`.	required

Returns:

Type	Description
`IndexStats`	Counts of the run.

Raises:

Type	Description
`ValueError`	if the indexer has no `dataset`.

`index_actions(actions)`

Index a stream of already formatted actions.

Parameters:

Name	Type	Description	Default
`actions`	`Actions`	Bulk actions, e.g. from `format_entities`.	required

Returns:

Type	Description
`IndexStats`	Counts of the run.

`openaleph_search.index.indexer.IndexStats`

Bases: NamedTuple

Outcome of an indexing run.

Attributes:

Name	Type	Description
`indexed`	`int`	Documents indexed.
`failed`	`int`	Documents rejected for good.
`took`	`timedelta`	Duration of the run.

Read

`openaleph_search.index.entities.get_entity(entity_id, **kwargs)`

Fetch an entity from the index.

Parameters:

Name	Type	Description	Default
`entity_id`		The entity ID.	required
`**kwargs`		Passed to `entities_by_ids`.	`{}`

Returns:

Type	Description
	The entity, or `None` if it is not indexed.

`openaleph_search.index.entities.entities_by_ids(ids, schemata=None, cached=False, includes=PROXY_INCLUDES, excludes=None)`

Iterate over the entities with the given IDs.

Parameters:

Name	Description	Default
`ids`	One or more entity IDs.	required
`schemata`	Only look in the buckets of these schemata.	`None`
`cached`	Not implemented, logs a warning.	`False`
`includes`	`_source` fields to return.	`PROXY_INCLUDES`
`excludes`	`_source` fields to leave out.	`None`

Yields:

Type	Description
	Entities as `unpack_result` returns them.

`openaleph_search.index.entities.iter_entities(auth=None, dataset=None, collection_id=None, schemata=None, includes=PROXY_INCLUDES, excludes=None, filters=None, sort=None, es_scroll='5m', es_scroll_size=1000)`

Scan all entities matching the given criteria.

Parameters:

Name	Type	Description	Default
`auth`	`SearchAuth \| None`	Only the datasets it may read.	`None`
`dataset`	`str \| None`	Only this dataset; ignored with `collection_id`.	`None`
`collection_id`	`str \| None`	Only this collection.	`None`
`schemata`		Only entities of these schemata or their descendants.	`None`
`includes`		`_source` fields to return.	`PROXY_INCLUDES`
`excludes`		`_source` fields to leave out.	`None`
`filters`		More Elasticsearch filter clauses.	`None`
`sort`		Sort to scan in; unsorted scans are faster.	`None`
`es_scroll`		Lifetime of the scroll context.	`'5m'`
`es_scroll_size`		Hits per scroll page.	`1000`

Yields:

Type	Description
	Entities as `unpack_result` returns them.

`openaleph_search.index.entities.get_entity_content(entity_id)`

Fetch the indexed fulltext of a Document or Pages entity: its bodyText, or the stored content of the pages bucket.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	The entity ID.	required

Returns:

Type	Description
`str \| None`	The text joined by newlines, or `None` if the entity is not indexed, not a `Document` (`Page` is not) or has no text.

Transform

`openaleph_search.transform.entity.format_entity(dataset, entity, **kwargs)`

Transform an entity into a bulk index action. Trusted entity data (an EntityPayload or a dict) is transformed without building a proxy; the input is not modified.

Parameters:

Name	Type	Description	Default
`dataset`	`str`	Dataset of the entity.	required
`entity`	`EntityLike`	A proxy, an `EntityPayload` or a trusted entity dict.	required
`**kwargs`		`collection_id` to store (deprecated).	`{}`

Returns:

Type	Description
`Action \| None`	The action, or `None` for an entity of an abstract schema.

`openaleph_search.transform.entity.format_entities(dataset, entities, **kwargs)`

Lazily transform a stream of entities into index actions.

Parameters:

Name	Type	Description	Default
`dataset`	`str`	Dataset of the entities.	required
`entities`	`Iterable[EntityLike]`	Proxies, `EntityPayload`s or trusted entity dicts.	required
`**kwargs`		Passed to `format_entity`.	`{}`

Yields:

Type	Description
`Actions`	Actions, skipping entities of abstract schemata.

`openaleph_search.transform.entity.iter_batches(entities, chunk_size=None, batch_bytes=None)`

Batch a stream of entities by count and by payload bytes (estimated by entity_size), cutting on whichever bound is reached first.

Parameters:

Name	Type	Description	Default
`entities`	`Iterable[EntityLike]`	Proxies, `EntityPayload`s or trusted entity dicts.	required
`chunk_size`	`int \| None`	Entities per batch, defaults to `indexer_chunk_size`.	`None`
`batch_bytes`	`int \| None`	Payload bytes per batch, defaults to `indexer_batch_bytes`.	`None`

Yields:

Type	Description
`list[EntityLike]`	Lists of entities.