swh.indexer.metadata module

swh.indexer.metadata.call_with_batches(f: Callable[[List[T1]], Iterable[T2]], args: List[T1], batch_size: int) → Iterator[T2]

Calls a function with batches of args, and concatenates the results.

class swh.indexer.metadata.ExtrinsicMetadataIndexer(config=None, **kw)

Bases: BaseIndexer[bytes, RawExtrinsicMetadata, OriginExtrinsicMetadataRow]

Prepare and check that the indexer is ready to run.

process_journal_objects(objects: ObjectsDict) → Dict

Read swh message objects (content, origin, …) from the journal to:

• retrieve the associated objects from the storage backend (e.g. storage, objstorage…)

• execute the associated indexing computations

• store the results in the indexer storage

index(id: bytes, data: RawExtrinsicMetadata | None, **kwargs) → List[OriginExtrinsicMetadataRow]

Index computation for the id and associated raw data.

Parameters:

• id – identifier or Dict object

• data – id’s data from storage or objstorage depending on object type

Returns:

a dict that makes sense for the persist_index_computations() method.

Return type:

dict

persist_index_computations(results: List[OriginExtrinsicMetadataRow]) → Dict[str, int]

Persist the results in storage.

class swh.indexer.metadata.ContentMetadataIndexer(config=None, **kw)

Bases: ContentIndexer[ContentMetadataRow]

Content-level indexer

This indexer is in charge of:

• filtering out content already indexed in content_metadata

• reading content from objstorage with the content’s id sha1

• computing metadata by given context

• using the metadata_dictionary as the ‘swh-metadata-translator’ tool

• store result in content_metadata table

Prepare and check that the indexer is ready to run.

filter(ids: List[CompositeObjId])

Filter out known sha1s and return only missing ones.

index(id: CompositeObjId, data: bytes | None = None, log_suffix='unknown directory', **kwargs) → List[ContentMetadataRow]

Index sha1s’ content and store result.

Parameters:

• id – content’s identifier

• data – raw content in bytes

Returns:

dictionary representing a content_metadata. If the translation wasn’t successful the metadata keys will be returned as None

Return type:

dict

persist_index_computations(results: List[ContentMetadataRow]) → Dict[str, int]

Persist the results in storage.

class swh.indexer.metadata.DirectoryMetadataIndexer(*args, **kwargs)

Bases: DirectoryIndexer[DirectoryIntrinsicMetadataRow]

Directory-level indexer

This indexer is in charge of:

• filtering directories already indexed in directory_intrinsic_metadata table with defined computation tool

• retrieve all entry_files in directory

• use metadata_detector for file_names containing metadata

• compute metadata translation if necessary and possible (depends on tool)

• send sha1s to content indexing if possible

• store the results for directory

Prepare and check that the indexer is ready to run.

filter(sha1_gits)

Filter out known sha1s and return only missing ones.

index(id: bytes, data: Directory | None = None, **kwargs) → List[DirectoryIntrinsicMetadataRow]

Index directory by processing it and organizing result.

use metadata_detector to iterate on filenames, passes them to the content indexers, then merges (if more than one)

Parameters:

• id – sha1_git of the directory

• data – should always be None

Returns:

dictionary representing a directory_intrinsic_metadata, with keys:

• id: directory’s identifier (sha1_git)

• indexer_configuration_id (bytes): tool used

• metadata: dict of retrieved metadata

Return type:

dict

persist_index_computations(results: List[DirectoryIntrinsicMetadataRow]) → Dict[str, int]

Persist the results in storage.

translate_directory_intrinsic_metadata(files: List[DirectoryLsEntry], log_suffix: str) → Tuple[List[Any], Any]

Determine plan of action to translate metadata in the given root directory

Parameters:

files – list of file entries, as returned by swh.storage.interface.StorageInterface.directory_ls()

Returns:

list of mappings used and dict with translated metadata according to the CodeMeta vocabulary

Return type:

(List[str], dict)

class swh.indexer.metadata.OriginMetadataIndexer(config=None, **kwargs)

Bases: OriginIndexer[Tuple[OriginIntrinsicMetadataRow, DirectoryIntrinsicMetadataRow]]

Prepare and check that the indexer is ready to run.

USE_TOOLS = False

index_list(origins: List[Origin], *, check_origin_known: bool = True, **kwargs) → List[Tuple[OriginIntrinsicMetadataRow, DirectoryIntrinsicMetadataRow]]

persist_index_computations(results: List[Tuple[OriginIntrinsicMetadataRow, DirectoryIntrinsicMetadataRow]]) → Dict[str, int]

Persist the computation resulting from the index.

Parameters:

results – List of results. One result is the result of the index function.

Returns:

a summary dict of what has been inserted in the storage