biothings.hub.dataindex
biothings.hub.dataindex.idcache
biothings.hub.dataindex.indexer_cleanup
biothings.hub.dataindex.indexer_payload
- class biothings.hub.dataindex.indexer_payload.IndexMappings(dict=None, /, **kwargs)[source]
Bases:
_IndexPayload
biothings.hub.dataindex.indexer
- class biothings.hub.dataindex.indexer.ColdHotIndexer(build_doc, indexer_env, index_name)[source]
Bases:
objectMongoDB to Elasticsearch 2-pass Indexer. (
1st pass: <MongoDB Cold Collection>, # static data 2nd pass: <MongoDB Hot Collection> # changing data
- ) =>
<Elasticsearch Index>
- class biothings.hub.dataindex.indexer.DynamicIndexerFactory(urls, es_host, suffix='_current')[source]
Bases:
objectIn the context of autohub/standalone instances, create indexer with parameters taken from versions.json URL. A list of URLs is provided so the factory knows how to create these indexers for each URLs. There’s no way to “guess” an ES host from a URL, so this parameter must be specified as well, common to all URLs “suffix” param is added at the end of index names.
- class biothings.hub.dataindex.indexer.IndexManager(job_manager: JobManager, poll_schedule=None)[source]
Bases:
BaseManagerAn example of config dict for this module. {
- “indexer_select”: {
None: “hub.dataindex.indexer.DrugIndexer”, # default “build_config.cold_collection” : “mv.ColdHotVariantIndexer”,
}, “env”: {
- “prod”: {
“host”: “localhost:9200”, “indexer”: {
- “args”: {
“request_timeout”: 300, “retry_on_timeout”: True, “max_retries”: 10,
}, “bulk”: {
“chunk_size”: 50 “raise_on_exception”: False
}, “concurrency”: 3
}, “index”: [
# for information only, only used in index_info {“index”: “mydrugs_current”, “doc_type”: “drug”}, {“index”: “mygene_current”, “doc_type”: “gene”}
],
}, “dev”: { … }
}
}
- clean_stale_status()[source]
During startup, search for action in progress which would have been interrupted and change the state to “canceled”. Ex: some donwloading processes could have been interrupted, at startup, “downloading” status should be changed to “canceled” so to reflect actual state on these datasources. This must be overriden in subclass.
- cleanup(env=None, keep=3, dryrun=True, **filters)[source]
Delete old indices except for the most recent ones.
Examples
>>> index_cleanup() >>> index_cleanup("production") >>> index_cleanup("local", build_config="demo") >>> index_cleanup("local", keep=0) >>> index_cleanup(_id="<elasticsearch_index>")
- get_indexes_by_name(index_name=None, env_name=None, limit=10)[source]
Accept an index_name and return a list of indexes get from all elasticsearch environments or from specific elasticsearch environment.
If index_name is blank, it will be return all indexes. limit can be used to specify how many indexes should be return.
The list of indexes will be like this: [
- {
“index_name”: “…”, “build_version”: “…”, “count”: 1000, “creation_date”: 1653468868933, “environment”: {
“name”: “env name”, “host”: “localhost:9200”,
}
},
]
- get_pinfo()[source]
Return dict containing information about the current process (used to report in the hub)
- index(indexer_env, build_name, index_name=None, ids=None, **kwargs)[source]
Trigger an index creation to index the collection build_name and create an index named index_name (or build_name if None). Optional list of IDs can be passed to index specific documents.
- update_metadata(indexer_env, index_name, build_name=None, _meta=None)[source]
- Update _meta field of the index mappings, basing on
the _meta value provided, including {}.
the _meta value of the build_name in src_build.
the _meta value of the build with the same name as the index.
Examples
update_metadata(“local”, “mynews_201228_vsdevjd”) update_metadata(“local”, “mynews_201228_vsdevjd”, _meta={}) update_metadata(“local”, “mynews_201228_vsdevjd”, _meta={“author”:”b”}) update_metadata(“local”, “mynews_201228_current”, “mynews_201228_vsdevjd”)
- class biothings.hub.dataindex.indexer.Indexer(build_doc: dict, indexer_env: dict, index_name: str)[source]
Bases:
objectMongoDB -> Elasticsearch Indexer.
- async index(job_manager, **kwargs)[source]
Build an Elasticsearch index (self.es_index_name) with data from MongoDB collection (self.mongo_collection_name).
“ids” can be passed to selectively index documents.
- “mode” can have the following values:
‘purge’: will delete an index if it exists.
‘resume’: will use an existing index and add missing documents.
‘merge’: will merge data to an existing index.
‘index’ (default): will create a new index.
- class biothings.hub.dataindex.indexer.IndexerCumulativeResult(dict=None, /, **kwargs)[source]
Bases:
_IndexerResult
- class biothings.hub.dataindex.indexer.IndexerStepResult(dict=None, /, **kwargs)[source]
Bases:
_IndexerResult
- class biothings.hub.dataindex.indexer.MainIndexStep(indexer)[source]
Bases:
Step- method: <property object at 0x77e7bbc5b650> = 'do_index'
- name: <property object at 0x77e7bbc5b560> = 'index'
- state
alias of
MainIndexJSR
- class biothings.hub.dataindex.indexer.PostIndexStep(indexer)[source]
Bases:
Step- method: <property object at 0x77e7bbc5b650> = 'post_index'
- name: <property object at 0x77e7bbc5b560> = 'post'
- state
alias of
PostIndexJSR
- class biothings.hub.dataindex.indexer.PreIndexStep(indexer)[source]
Bases:
Step- method: <property object at 0x77e7bbc5b650> = 'pre_index'
- name: <property object at 0x77e7bbc5b560> = 'pre'
- state
alias of
PreIndexJSR
- class biothings.hub.dataindex.indexer.ProcessInfo(indexer, concurrency)[source]
Bases:
object
- class biothings.hub.dataindex.indexer.Step(indexer)[source]
Bases:
ABC- catelog = {'index': <class 'biothings.hub.dataindex.indexer.MainIndexStep'>, 'post': <class 'biothings.hub.dataindex.indexer.PostIndexStep'>, 'pre': <class 'biothings.hub.dataindex.indexer.PreIndexStep'>}
- method: <property object at 0x77e7bbc5b650>
- name: <property object at 0x77e7bbc5b560>
- state: <property object at 0x77e7bbc5b4c0>
biothings.hub.dataindex.indexer_registrar
- class biothings.hub.dataindex.indexer_registrar.IndexJobStateRegistrar(collection, build_name, index_name, **context)[source]
Bases:
object
- class biothings.hub.dataindex.indexer_registrar.MainIndexJSR(collection, build_name, index_name, **context)[source]
Bases:
IndexJobStateRegistrar
- class biothings.hub.dataindex.indexer_registrar.PostIndexJSR(collection, build_name, index_name, **context)[source]
Bases:
IndexJobStateRegistrar
- class biothings.hub.dataindex.indexer_registrar.PreIndexJSR(collection, build_name, index_name, **context)[source]
Bases:
IndexJobStateRegistrar
biothings.hub.dataindex.indexer_schedule
- class biothings.hub.dataindex.indexer_schedule.Schedule(total: int, batch_size: int)[source]
Bases:
object
- exception biothings.hub.dataindex.indexer_schedule.SchedulerMismatchError(completed_documents: int, expected_documents: int)[source]
Bases:
ExceptionException for indicating a mismatch in the number of expected documents versus the total number of actually uploaded documents to the target database
biothings.hub.dataindex.indexer_task
- class biothings.hub.dataindex.indexer_task.ESIndex(client, index_name, logger=None, **bulk_index_args)[source]
Bases:
ESIndex- mexists(ids)[source]
Return a list of tuples like [
(_id_0, True), (_id_1, False), (_id_2, True), ….
]
- class biothings.hub.dataindex.indexer_task.IndexingTask(es, mongo, ids, mode=None, logger=None, name='task')[source]
Bases:
objectIndex one batch of documents from MongoDB to Elasticsearch. The documents to index are specified by their ids.
biothings.hub.dataindex.snapshooter
- class biothings.hub.dataindex.snapshooter.Bucket(client, bucket, region=None)[source]
Bases:
object
- class biothings.hub.dataindex.snapshooter.CloudStorage(type: str, access_key: str, secret_key: str, region: str = 'us-west-2')[source]
Bases:
object- access_key: str
- region: str = 'us-west-2'
- secret_key: str
- type: str
- class biothings.hub.dataindex.snapshooter.CumulativeResult(dict=None, /, **kwargs)[source]
Bases:
_SnapshotResult
- class biothings.hub.dataindex.snapshooter.ProcessInfo(env)[source]
Bases:
objectJobManager Process Info. Reported in Biothings Studio.
- class biothings.hub.dataindex.snapshooter.RepositoryConfig(dict=None, /, **kwargs)[source]
Bases:
UserDict- {
“type”: “s3”, “name”: “s3-$(Y)”, “settings”: {
“bucket”: “<SNAPSHOT_BUCKET_NAME>”, “base_path”: “mynews.info/$(Y)”, # per year
}
}
- property bucket
- format(doc=None)[source]
Template special values in this config.
For example: {
“bucket”: “backup-$(Y)”, “base_path” : “snapshots/%(_meta.build_version)s”
} where “_meta.build_version” value is taken from doc in dot field notation, and the current year replaces “$(Y)”.
- property region
- property repo
- class biothings.hub.dataindex.snapshooter.SnapshotEnv(job_manager, cloud, repository, indexer, **kwargs)[source]
Bases:
object
- class biothings.hub.dataindex.snapshooter.SnapshotManager(index_manager, *args, **kwargs)[source]
Bases:
BaseManagerHub ES Snapshot Management
Config Ex:
# env.<name>: {
- “cloud”: {
“type”: “aws”, # default, only one supported. “access_key”: <——————>, “secret_key”: <——————>, “region”: “us-west-2”
}, “repository”: {
“name”: “s3-$(Y)”, “type”: “s3”, “settings”: {
“bucket”: “<SNAPSHOT_BUCKET_NAME>”, “base_path”: “mygene.info/$(Y)”, # year
}, “acl”: “private”,
}, “indexer”: {
“name”: “local”, “args”: {
“request_timeout”: 100, “max_retries”: 5
}
}, “monitor_delay”: 15,
}
- clean_stale_status()[source]
During startup, search for action in progress which would have been interrupted and change the state to “canceled”. Ex: some donwloading processes could have been interrupted, at startup, “downloading” status should be changed to “canceled” so to reflect actual state on these datasources. This must be overriden in subclass.
- cleanup(env=None, keep=3, group_by='build_config', dryrun=True, ignoreErrors=False, **filters)[source]
Delete past snapshots and keep only the most recent ones.
Examples
>>> snapshot_cleanup() >>> snapshot_cleanup("s3_outbreak") >>> snapshot_cleanup("s3_outbreak", keep=0)
- delete_snapshot_from_db(build_name, snapshot_name)[source]
Delete a snapshot entry from the MongoDB database.
This method removes the specified snapshot from the build document in the source build collection. Called when a snapshot is found to be missing in the snapshot environment during validation.
- Parameters:
build_name (str) – The name of the build from which to delete the snapshot.
snapshot_name (str) – The name of the snapshot to delete.
- Returns:
None
- poll(state, func)[source]
Search for source in collection ‘col’ with a pending flag list containing ‘state’ and and call ‘func’ for each document found (with doc as only param)
- snapshot(snapshot_env, index, snapshot=None, recreate_repo=False)[source]
Create a snapshot named “snapshot” (or, by default, same name as the index) from “index” according to environment definition (repository, etc…) “env”.
- snapshot_a_build(build_doc)[source]
Create a snapshot basing on the autobuild settings in the build config. If the build config associated with this build has: {
- “autobuild”: {
“type”: “snapshot”, // implied when env is set. env must be set. “env”: “local” // which es env to make the snapshot.
} Attempt to make a snapshot for this build on the specified es env “local”.
- validate_snapshots()[source]
Validate the snapshots stored in the database.
This method checks each snapshot in the source build collection to verify whether it exists in the corresponding snapshot environment. If a snapshot does not exist in the environment, it is removed from the database. This helps keep the database in sync with the actual snapshots present in the environments.
The validation process is executed asynchronously and any errors encountered during validation are logged.
- Returns:
The asynchronous task representing the validation process.
- Return type:
job (asyncio.Task)
biothings.hub.dataindex.snapshot_cleanup
- biothings.hub.dataindex.snapshot_cleanup.delete(collection, element, envs, ignoreErrors=False)[source]
- biothings.hub.dataindex.snapshot_cleanup.find(collection, *, env=None, keep=3, group_by=None, return_db_cols=False, **filters)[source]
Identify snapshots to remove or keep based on specified criteria.
This function queries a MongoDB collection to find snapshots matching the given filters, groups them according to the specified grouping key(s), and determines which snapshots to keep or remove based on the ‘keep’ parameter.
Parameters: - collection (Collection): The MongoDB collection to query. Must be an instance of pymongo.collection.Collection. - env (str, optional): The environment name to filter snapshots. Defaults to None. - keep (int, optional): The number of most recent snapshots to keep in each group. Defaults to 3. - group_by (str or list, optional): The key or list of keys to group snapshots by. If None, defaults to ‘build_config’. - return_db_cols (bool, optional): If True, returns the raw database query results instead of the structured _Ele element. Defaults to False. - **filters: Additional keyword arguments to filter snapshots.
Returns: - _Ele or list: An _Ele element representing the snapshots to be removed and kept, organized by groups,
or a list of raw database query results if return_db_cols is True.
Raises: - NotImplementedError: If ‘collection’ is not an instance of pymongo.collection.Collection. - TypeError: If ‘group_by’ is neither a string, list, tuple, nor None.
biothings.hub.dataindex.snapshot_registrar
- class biothings.hub.dataindex.snapshot_registrar.MainSnapshotState(col, _id)[source]
Bases:
_TaskState- func = '_snapshot'
- name = 'snapshot'
- regx = True
- step = 'snapshot'
- class biothings.hub.dataindex.snapshot_registrar.PostSnapshotState(col, _id)[source]
Bases:
_TaskState- func = 'post_snapshot'
- name = 'post'
- regx = True
- step = 'post-snapshot'