mgnipy.V2.collect.biosampler module

mgnipy.V2.collect.biosampler module#

class BioSampler(sample_ids, config=None, metadata=None)[source]#

Bases: CheckpointMixin, ClientManagerMixin

Fetches BioSamples metadata for a given list of ENA run or sample accessions.

BioSampler is designed to retrieve the rich sample metadata from BioSamples for a list of Run or Sample ENA accessions.

It uses the get_biosample_metadata() or aget_biosample_metadata() (TODO) function to fetch the metadata for each accession with option to cache the results using the CheckpointMixin to avoid redundant API calls in future runs.

Parameters:

  • sample_ids (list of str ) – A list of ENA run or sample accessions for which to fetch the BioSamples metadata. If a run accession then enrich(incl_ena=True)() is required so that the sample accession can be retrieved from ENA first and then passed to a BioSamples API request.

  • config (MGnipyConfig, optional) – An optional configuration object for MGnipy. If not provided, a default configuration will be used.

  • client (Client or AuthenticatedClient, optional) – An optional HTTP client for making requests. If not provided, a default client will be initialized using the provided or default configuration.

  • metadata (ResultsHandler, optional) – An optional ResultsHandler instance to store the enriched metadata. If not provided, a new ResultsHandler will be created to hold the results.

all_ids#

The complete list of ENA run or sample accessions provided during initialization.

Type:

list of str

metadata#

The enriched metadata as a ResultsHandler instance.

Type:

ResultsHandler

Notes

  • The enrich() method iterates through the list of accessions and fetches their metadata, storing the results in a ResultsHandler instance.

  • By default there is the option to include ENA metadata in the enrichment process, which can be controlled via the incl_ena parameter. If incl_ena is False, then only sample accessions will return BioSamples metadata!

  • The aenrich() method is intended to provide an asynchronous version of the enrichment process, but it is currently not implemented. Future updates will include asynchronous fetching of metadata to improve performance for large datasets.

  • The class is designed to be flexible, allowing users to specify a limit on the number of accessions to enrich in a single run, which is useful for testing or when dealing with large datasets to avoid long runtimes during development. If the limit is set to None, there will be no limit on the number of accessions enriched.

  • It uses the CheckpointMixin to cache results and avoid redundant API calls.

async aclose()#
async aenrich(limit=200, hide_progress=False, incl_ena=False, skip_failed=False)[source]#

Async version of enrich().

See enrich() for details on parameters and behavior.

This is a placeholder and not yet implemented.

Parameters:
Return type:

None

property all_ids: list [str ]#

The list of ENA run or sample accessions set during initialization.

async aload_cache()#

Async wrapper for load_cache.

Return type:

list [int ]

property async_httpx_client: AsyncClient#

Get the asynchronous httpx client instance from the AuthenticatedClient.

Returns:

The asynchronous httpx client instance.

Return type:

httpx.AsyncClient

async awrite_results(request_num, items)#

Async wrapper for write_results.

Parameters:
Return type:

None

property cache_key: str #

Generate deterministic hash from resource + params.

Returns:

A unique cache key for the current query parameters and resource. For a query to the ‘samples’ resource with parameters {‘biome_lineage’: ‘root:Environmental:Terrestrial’}, the cache key will be a SHA256 hash of the string representation of the resource and parameters, ensuring that identical queries will have the same cache key and thus access the same cached results.

Return type:

str

Example

>>> from mgnipy.V2.mixins import CheckpointMixin
>>> from mgnipy import MGnipyConfig
>>> # Prepare parameters and config
>>> params = {'lineage': 'root:Environmental:Terrestrial'}
>>> resource = 'biome'
>>> config = MGnipyConfig(cache_dir="/path/to/cache")
>>> # Create CheckpointMixin instance and compute cache key
>>> cache_handler = CheckpointMixin()
>>> cache_handler.params = params
>>> cache_handler.resource = resource
>>> cache_handler.config = config
>>> cache_handler.cache_key
'1eb56ddf5a2e7d60d8155c8bbe01f032f959a2519d43e99f31f533abffa3166f'
property cache_path: Path | None #

Directory for this query’s cached pages.

clear_cache()#

Remove all cached pages for this set of queries.

Return type:

None

close()#
enrich(limit=200, hide_progress=False, incl_ena=False, skip_failed=True)[source]#

Fetches BioSample metadata for the given run/sample accessions.

This method iterates through the list of ENA run or sample accessions provided during initialization and retrieves their corresponding BioSample metadata. The results are stored in the ResultsHandler instance associated with this class. This does not return anything.

Parameters:

  • limit (Optional[int ], default=200) – An optional integer to limit the number of biosamples to enrich. If set to None, there will be no limit on the number of biosamples enriched.

  • hide_progress (bool , default=False) – Whether to hide the progress bar during enrichment.

  • incl_ena (bool , default=False) – Whether to include an API call to ENA prior to the BioSamples requeßst. If set to False, only sample accessions will return BioSamples metadata.

  • skip_failed (bool , default=True) – Whether to skip failed enrichments. If set to True, failed enrichments will be logged and skipped, and a placeholder with the GivenID will be appended to the results (appear as completed in ResultsHandler.get_ids()). If set to False, any failed enrichments will not be appended to the results (appear as still left to do).

property httpx_client: Client#

Get the synchronous httpx client instance from the AuthenticatedClient.

Returns:

The synchronous httpx client instance.

Return type:

httpx.Client

load_cache()#

Pick up where you left off. Loads cached results and manifest into memory.

Returns:

A list of request numbers (page numbers) that were successfully loaded from the cache.

Return type:

list of int

load_cache_manifest()#

Load the cache manifest file if present, and update total records and total requests.

Returns:

The contents of the manifest file, or an empty dictionary if the manifest is not found or fails to load.

Return type:

dict

load_cache_results()#

Load cached pages/request nums into results.

Loads cached pages from disk into the in-memory results dictionary (self._results), if available.

Returns:

A list of request numbers (page numbers) that were successfully loaded from the cache.

Return type:

list of int

property manifest_path: Path | None #

Path to mgnipy_manifest.json storing metadata.

property metadata: MGnifyMetadata#

The enriched metadata as a MGnifyMetadata instance.

property params: dict [str , Any ]#

For CheckpointMixin

renew_client()#

Init a new client instance and replace the existing one. This is useful if the current client has been closed or is no longer valid, allowing for a fresh start with a new HTTP client session.

property resource: str #

For CheckpointMixin

status()#

Print the status of the MGnipy client, including the type of client and whether the synchronous and asynchronous httpx client sessions are open.

Return type:

None

try_load_cache()#

Attempt to load cached results and manifest into memory if not already loaded. This method checks if the cache has already been loaded to avoid redundant operations. If the cache has not been loaded, it will attempt to load it and set the _cache_loaded attribute accordingly.

Notes

  • This method is intended to be called internally before accessing cached results.

  • If cache_dir is None then _cache_loaded will be True after initial attempt.

  • If an error occurs during cache loading, it will be logged, and _cache_loaded will be set to False.

  • Dependent on .mixins.CheckpointMixin

Return type:

None

write_results(request_num, items, include_manifest=True)#

Auto atomic write to disk.

Parameters:
Return type:

None

Contents