mgnipy.V2.mixins module

mgnipy.V2.mixins module#

class BiomesTreeMixin[source]#

Bases: object

property lineages: list [str ]#
property results: dict #

Get results and auto-normalize lineage field.

show_tree(method='compact')[source]#
Parameters:

method (Literal ['compact', 'show', 'print', 'horizontal', 'hshow', 'h', 'hprint', 'vertical', 'vshow', 'v', 'vprint'])

property tree: Tree#

Convert the biomes metadata to a tree structure for visualization or analysis.

Returns:

A tree representation of the biomes and their relationships.

Return type:

Tree

class CheckpointMixin[source]#

Bases: object

Checkpoint manager for request-makers.

This mixin provides methods to write paginated results to disk as they are retrieved, and to load them back into memory on subsequent runs. It generates a unique cache key based on the query parameters and resource type, and organizes cached pages in a directory structure under a specified cache root.

The mixin assumes the host class has the following attributes: - self.params: A dictionary of query parameters for the current request. - self.config: An instance of MGnipyConfig containing configuration settings, including the cache directory. - self.resource: A string representing the type of resource being queried (e.g., “samples”, “runs”). - self._results: A dictionary mapping page numbers to lists of dictionaries containing the query results. - self.count: An integer representing the total number of records for the current query. - self.num_requests: An integer representing the total number of paginated requests needed for the current query. which is the case for: QuerySets, MGnifier, proxies

Example

>>> from mgnipy.V2.mgnifier.query_set import QuerySet
>>> from mgnipy.V2.mixins import CheckpointMixin
>>> # Creating a class that uses mixin
>>> class MyQuerySet(QuerySet, CheckpointMixin):
...     # Initialize QuerySet (not mixin)
...     def __init__(self, *args, **kwargs):
...         super().__init__(*args, **kwargs)
>>> # init
>>> qs = MyQuerySet(resource="studies") # default config is w/ cache enabled
>>> # now can use checkpoint methods, e.g.
>>> qs.cache_key # print the unique cache key
'...'
>>> qs.load_cache()
[]
async aload_cache()[source]#

Async wrapper for load_cache.

Return type:

list [int ]

async awrite_results(request_num, items)[source]#

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()[source]#

Remove all cached pages for this set of queries.

Return type:

None

load_cache()[source]#

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()[source]#

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()[source]#

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.

try_load_cache()[source]#

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)[source]#

Auto atomic write to disk.

Parameters:
Return type:

None

class ClientManagerMixin[source]#

Bases: object

Context manager mixin for managing the lifecycle of an HTTP client (synchronous or asynchronous) within a class. This mixin provides methods to enter and exit the context, ensuring that the client is properly initialized and closed, and handles cases where the client may have been closed or is not available.

Requirements#

  • self.client must be an instance of a class that implements the context manager protocol (i.e., has __enter__ and __exit__ methods for synchronous clients, and __aenter__ and __aexit__ methods for asynchronous clients).

  • self.config must be a configuration object or dictionary that can be used to initialize a new HTTP client instance if the existing client is closed or unavailable.

async aclose()[source]#
property async_httpx_client: AsyncClient#

Get the asynchronous httpx client instance from the AuthenticatedClient.

Returns:

The asynchronous httpx client instance.

Return type:

httpx.AsyncClient

close()[source]#
property httpx_client: Client#

Get the synchronous httpx client instance from the AuthenticatedClient.

Returns:

The synchronous httpx client instance.

Return type:

httpx.Client

renew_client()[source]#

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.

status()[source]#

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

class StreamMixin[source]#

Bases: object

Mixin providing streaming helpers for downloads.

# TODO remove below dependencies on mgnifier This mixin assumes the host class provides the following helpers/properties:

.exec.httpx_client and .exec.httpx_aclient attributes

  • _get_type_by_alias(alias) to resolve file types

  • downloads_df when needed for examples/tests

The implementation mirrors the streaming helpers previously defined on MGazine so they can be reused by other classes.

stream(*, alias=None, url=None, chunksize=None, max_skip=5, **kwargs)[source]#

Streams a single download based on its alias or url.

If chunksize is specified then iterators of dataframes or strings will be returned; otherwise the full data will be returned as a single object.

Supported formats and their handlers#

param alias:

The alias of the download to stream.

type alias:

Optional[str]

param url:

The url of the download to stream.

type url:

Optional[HttpUrl]

param chunksize:

The size of the chunks to read from the stream.

type chunksize:

Optional[int]

param max_skip:

The maximum number of rows to skip before raising an error. Default is 5.

type max_skip:

int, optional

param **kwargs:

Additional keyword arguments to pass to the streamer function.

returns:

The streamer result for the resolved alias or url.

rtype:

Any

Parameters:

  • alias (str | None)

  • url (HttpUrl | None)

  • chunksize (int | None)

  • max_skip (int )

Return type:

Any

stream_biom(url, **skbio_kwargs)[source]#

Stream a biom file from a URL using scikit-bio’s read function. Refer there for more info.

Parameters:

  • url (str ) – The URL to the biom file to stream.

  • **skbio_kwargs – Additional keyword arguments passed to skbio.io.read(), such as into and verify.

Returns:

A generator yielding scikit-bio Sequence objects parsed from the biom file.

Return type:

Generator

stream_fasta(url, **skbio_kwargs)[source]#

Stream a FASTA file from a URL using scikit-bio’s read function. Refer there for more info.

Parameters:

  • url (str ) – The URL to the FASTA file to stream.

  • **skbio_kwargs – Additional keyword arguments passed to skbio.io.read(), such as into and verify.

Returns:

A generator yielding scikit-bio Sequence objects parsed from the FASTA file.

Return type:

Generator

stream_gff(url, **skbio_kwargs)[source]#

Stream a GFF file from a URL using scikit-bio’s read function. Refer there for more info.

Parameters:

  • url (str ) – The URL to the GFF file to stream.

  • **skbio_kwargs – Additional keyword arguments passed to skbio.io.read(), such as into and verify.

Returns:

A generator yielding scikit-bio Sequence objects parsed from the GFF file.

Return type:

Generator

stream_gzipped(url, chunksize=None, decode=False, encoding='utf-8', errors='replace', **httpx_kwargs)[source]#

Stream a gzipped HTTP resource and present a file-like interface.

When chunksize is None the entire compressed payload is fetched and decompressed into memory. When chunksize is provided a streaming file-like object is returned.

Parameters:
Return type:

bytes | str | BufferedReader | TextIOWrapper

stream_html(url, **web_kwargs)[source]#

Open an HTML URL in the default web browser.

Parameters:

  • url (str ) – The URL to open in the web browser.

  • **web_kwargs – Additional keyword arguments passed to webbrowser.open(), such as new and autoraise.

Returns:

True if the URL was opened successfully, False otherwise.

Return type:

bool

stream_json(url, chunksize=None, **httpx_kwargs)[source]#
Parameters:
Return type:

dict | Generator

stream_jsonl(url, orient=None, chunksize=None, df_engine='pandas', **df_kwargs)[source]#
Parameters:

  • url (str )

  • orient (Literal ['records', 'split', 'index', 'columns', 'values', 'table'] | None)

  • chunksize (int | None)

  • df_engine (Literal ['pandas', 'polars'] | None)

Return type:

dict

stream_pandas(url, sep='\t', chunksize=None, max_skip=5, low_memory=False, **pd_kwargs)[source]#

Read a TSV from a URL or local file with resilient header handling.

The helper will retry with increasing skiprows when pandas raises a ParserError (useful for files with extra header lines). When chunksize is provided an iterator is returned.

Parameters:

  • url (str ) – The URL or local file path to read the TSV from.

  • sep (str ) – The delimiter to use (default is tab).

  • chunksize (int or None) – If an integer is provided, returns an iterator that yields DataFrames of that many rows. If None, returns a single DataFrame.

  • max_skip (int ) – The maximum number of lines to skip when trying to parse the TSV.

  • **pd_kwargs – Additional keyword arguments passed to pd.read_csv.

  • low_memory (bool )

Returns:

A DataFrame containing the TSV data, or an iterator yielding DataFrames if chunksize is specified.

Return type:

pd.DataFrame or TextFileReader

Raises:

  • ValueError – If chunksize is not a positive integer or None.

  • RuntimeError – If the TSV cannot be parsed after skipping up to max_skip lines.

  • Pandas ParserError – If the TSV cannot be parsed due to a format error (after retries).

stream_polars(url, sep='\t', chunksize=None, max_skip=5, low_memory=False, **pl_kwargs)[source]#

Read a TSV from a URL or local file into a Polars DataFrame with resilient header handling.

The helper will retry with increasing skip_rows when Polars raises an error (useful for files with extra header lines). When chunksize is provided an iterator is returned.

Parameters:

  • url (str ) – The URL or local file path to read the TSV from.

  • sep (str ) – The delimiter to use (default is tab).

  • chunksize (int or None) – If an integer is provided, returns an iterator that yields DataFrames of that many rows. If None, returns a single DataFrame.

  • max_skip (int ) – The maximum number of lines to skip when trying to parse the TSV.

  • **pl_kwargs – Additional keyword arguments passed to pl.read_csv.

  • low_memory (bool )

Returns:

A Polars DataFrame containing the TSV data, or an iterator yielding DataFrames if chunksize is specified.

Return type:

pl.DataFrame or Iterator[pl.DataFrame]

Raises:

  • ValueError – If chunksize is not a positive integer or None.

  • RuntimeError – If the TSV cannot be parsed after skipping up to max_skip lines.

  • Polars Error – If the TSV cannot be parsed due to a format error (after retries).

stream_tree(url, **skbio_kwargs)[source]#
Parameters:

url (str )

Return type:

Generator

stream_txt(url, chunksize=None, **httpx_kwargs)[source]#

Stream a plain-text resource. When chunksize is None the full text is returned as a string. When chunksize is an integer the function yields lists of lines.

Parameters:

  • url (str ) – The URL to stream the text from.

  • chunksize (int or None) – If an integer is provided, yields lists of lines of that size. If None, yields the entire text as a single string.

  • httpx_client (httpx.Client, optional) – An optional httpx.Client to use for the request. If None, a new client will be created for the request.

  • **httpx_kwargs – Additional keyword arguments passed to the httpx.Client.request() method

Returns:

The full text as a string if chunksize is None, or a generator yielding lists of lines if chunksize is an integer.

Return type:

str or Generator

Contents