mgnipy package

mgnipy package#

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:

limit (int | None)
hide_progress (bool )
incl_ena (bool )
skip_failed (bool )

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:

request_num (int )
items (Any )

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:

request_num (int )
items (Any )
include_manifest (bool )

Return type:

None

class MGazine(downloads, config=None, *, client=None, mgnify_studies=None, mgnify_analyses=None, mgnify_runs=None, mgnify_samples=None, mgnify_assemblies=None, biosamples_metadata=None, obs=None)[source]#

Bases: StreamMixin, ClientManagerMixin, MetadataSettersMixin

Reads or downloads datasets from MGnify.

MGazine is a class for managing and downloading datasets from MGnify. - Accepts a list of download-like dictionaries (for example the objects returned by the MGnify API for downloads) and provides simple streaming and download helpers. - Supports grouping datasets by pipeline version and short description, and provides methods for downloading individual files or all files in the MGazine.

Parameters:

downloads (list of dict ) – A list of download-like dictionaries, each containing keys such as alias, url, file_type, download_group, short_description, and pipeline_version.
config (MGnipyConfig, optional) – An optional configuration object for MGnipy. If not provided, a default configuration is used.
client (Client or AuthenticatedClient, optional) – An optional client object for making HTTP requests. If not provided, a default client is used.
mgnify_[studies|analyses|runs|samples|assemblies] (list of dict , optional) – Lists of dictionaries containing metadata for each respective MGnify dataset.
biosamples_metadata (list of dict , optional) – A list of dictionaries containing metadata for BioSamples.
mgnify_studies (list [dict [str , Any]] | None)
mgnify_analyses (list [dict [str , Any]] | None)
mgnify_runs (list [dict [str , Any]] | None)
mgnify_samples (list [dict [str , Any]] | None)
mgnify_assemblies (list [dict [str , Any]] | None)
obs (list [dict [str , Any]] | None)

downloads#

The list of download-like dictionaries provided during initialization.

Type:: list of dict

downloads_df[source]#

A DataFrame representation of the downloads, with columns such as alias, url, and file_type.

Type:: pandas.DataFrame
Return type:: DataFrame

aliases#

A list of all download aliases extracted from the downloads.

Type:: list of str

urls#

An alias for url_list, providing a list of all download URLs.

Type:: list of str

url_list#

A list of URLs extracted from the downloads.

Type:: list of str

url_dict#

A dictionary mapping each download alias to its corresponding URL.

Type:: dict

lazy_merged#

A lazy frame containing the merged datasets, if initialized.

Type:: polars.LazyFrame or None

short_desc#

The short description of the MGazine, derived from the downloads. If multiple short descriptions are present, a warning is issued.

Type:: str

Example

>>> downloads = [
...    {"alias": "a", "url": "/tmp/a.txt", "file_type": "txt", "short_description": "desc1", "pipeline_version": "v5"},
...    {"alias": "boop", "url": "/tmp/b.fasta", "file_type": "fasta", "short_description": "desc2", "pipeline_version": "v5"},
... ]
>>> mg = MGazine(downloads)
>>> print(mg)
MGazine containing:
- MGnify pipeline versions: ['v5']
- Number of downloads: 2
- Short descriptions: ['desc1', 'desc2']
- Nonempty metadata sets:

async aclose()#

async adownload(to_dir, alias=None, *, url=None, filename=None, overwrite=False, hide_progress=False)[source]#

Asynchronously download a file from an alias or URL.

Parameters:

to_dir (DirectoryPath) – Directory where the file will be saved.
alias (str or None, optional) – Download alias known to this MGazine instance.
url (str or None, optional) – Direct URL to fetch. Either alias or url must be provided.
filename (str or None, optional) – Filename to use for the saved file. When omitted the alias is used.
httpx_aclient (httpx.AsyncClient, optional) – Optional httpx.AsyncClient to use for the HTTP request.
overwrite (bool , optional) – If False and the destination file already exists the download is skipped. When True the existing file will be overwritten.
hide_progress (bool , optional) – Disable the progress bar when True.

Raises:

ValueError – If neither alias nor url is provided.

Examples

downloads = [ … { … “alias”: “example.txt”, … “url”: “http://ex/x ”, … “file_type”: “txt”, … }] mg = MGazine(downloads) await mg.adownload(“download_to_here”, alias=”example.txt”) # doctest: +SKIP

async adownload_all(to_dir, overwrite=False, hide_progress=False)[source]#

Asynchronously download all files known to this MGazine.

Parameters:

to_dir (DirectoryPath) – Directory where the files will be saved.
overwrite (bool , optional) – Passed to adownload to control overwriting behavior.
hide_progress (bool , optional) – Disable progress bars when True.

Notes

This helper creates a single async HTTP client and schedules concurrent adownload calls for all aliases.

Examples

>>> downloads = [
...     {"alias": "example.txt", "url": "http://ex/x", "file_type": "txt"},
...     {"alias": "example2.fasta.gz", "url": "http://ex/x2", "file_type": "fasta"},
... ]
>>> mg = MGazine(downloads)
>>> await mg.adownload_all("download_to_here")

property aliases: list [str ]#

Return a list of all download aliases.

Example

>>> downloads = [{"alias": "example.txt", "url": "http://ex/x"}]
>>> MGazine(downloads).aliases
['example.txt']

append_biosamples_metadata(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_analyses(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_assemblies(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_runs(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_samples(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_studies(value)#

Parameters:: value (dict [str , Any ])

append_obs(value)#

Parameters:: value (dict [str , Any ])

property async_httpx_client: AsyncClient#

Get the asynchronous httpx client instance from the AuthenticatedClient.

Returns:: The asynchronous httpx client instance.
Return type:: httpx.AsyncClient

property available_metadata_sets: list [str ]#

Return a list of available metadata sets in the MGazine.

This property checks which metadata sets (e.g., studies, analyses, runs, samples, assemblies, biosamples) are non-empty and returns their names as a list.

Returns:: A list of names of non-empty metadata sets available in the MGazine.
Return type:: list of str

Examples

>>> mg = MGazine(downloads)
>>> mg.available_metadata_sets
['mgnify_studies', 'mgnify_analyses', 'mgnify_runs']

property biosamples_metadata: ResultsHandler#

by_downloads_col(col)[source]#

Group downloads by a specified column in the downloads dataframe.

Parameters:: col (str ) – The column name to group by.
Returns:: A dictionary where keys are unique values from the specified column and values are lists of download dictionaries.
Return type:: dict
Raises:: ValueError – If the specified column is not present in the downloads dataframe.

close()#

download(to_dir, alias=None, *, url=None, filename=None, overwrite=False, hide_progress=False)[source]#

Download a file by its alias or URL.

Download a file from an alias or URL to a local directory.

Parameters:

to_dir (DirectoryPath) – Directory where the file will be saved.
alias (str or None, optional) – Download alias known to this MGazine instance. When provided the corresponding URL from the instance’s downloads list is used.
url (str or None, optional) – Direct URL to fetch. Either alias or url must be provided.
filename (str or None, optional) – Filename to use for the saved file. When omitted the alias is used.
overwrite (bool , optional) – If False and the destination file already exists the download is skipped. When True the existing file will be overwritten.
hide_progress (bool , optional) – Disable the progress bar when True.

Raises:

ValueError – If neither alias nor url is provided.

Examples

mg = MGazine(downloads) # doctest: +SKIP mg.download(“download_to_here”, alias=”example.txt”) # doctest: +SKIP

download_all(to_dir, hide_progress=False, overwrite=False)[source]#

Download all files known to this MGazine instance.

Parameters:

to_dir (DirectoryPath) – Directory where the files will be saved.
hide_progress (bool , optional) – Disable per-file and overall progress bars when True.
overwrite (bool , optional) – Passed to download to control overwriting behavior.

Notes

This helper calls download for each alias present in the instance’s downloads list.

Examples

>>> downloads = [
...     {"alias": "example.txt", "url": "http://ex/x", "file_type": "txt"},
...     {"alias": "example2.fasta.gz", "url": "http://ex/x2", "file_type": "fasta"},
... ]
>>> mg = MGazine(downloads)
>>> mg.download_all("download_to_here")

downloads_df(**pd_kwargs)[source]#

The downloads as a DataFrame.

This returns a pandas.DataFrame of all downloads. The dataframe should contain columns such as alias, url and file_type (TODO pandera).

Parameters:: pd_kwargs (dict ) – Additional keyword arguments to pass to the pandas.DataFrame constructor.
Returns:: A DataFrame containing the downloads information
Return type:: pd.DataFrame

Examples

>>> downloads = [{"alias": "example.txt", "url": "http://ex/x", "file_type": "txt"}]
>>> mag = MGazine(downloads)
>>> df = mag.downloads_df(index=["boop"])

property httpx_client: Client#

Get the synchronous httpx client instance from the AuthenticatedClient.

Returns:: The synchronous httpx client instance.
Return type:: httpx.Client

lazy_concat(aliases=None, urls=None, how='vertical_relaxed', **pl_kwargs)[source]#

Return a concatenated Polars LazyFrame of the datasets corresponding to the provided aliases or URLs.

Parameters:

aliases (list [str ] or None, optional) – List of download aliases to stream and concatenate. If provided, this takes precedence over urls.
urls (list [str ] or None, optional) – List of download URLs to stream and concatenate. Used only if aliases is not provided.
how (str , optional) – Concatenation method. Options include ‘vertical’, ‘horizontal’, ‘vertical_relaxed’, etc. See Polars documentation for details.
**pl_kwargs – Additional keyword arguments to pass to the Polars concatenation function.

Returns:

A Polars LazyFrame representing the concatenated datasets.

Return type:

pl.LazyFrame

property lazy_merged: LazyFrame | None #

Return the current lazy merged Polars LazyFrame if available.

Returns:: The current lazy merged Polars LazyFrame, or None if not set.
Return type:: pl.LazyFrame or None

list_pipeline_version()[source]#

A list of unique pipeline versions in the MGazine.

Returns:: A list of unique pipeline versions extracted from the downloads.
Return type:: list of str

Examples

>>> downloads = [
...     {"alias": "example.txt", "url": "http://ex/x", "pipeline_version": 'v4_1'},
...     {"alias": "example2.txt", "url": "http://ex/x2", "pipeline_version": 'v5'},
... ]
>>> MGazine(downloads).list_pipeline_version()
['v4_1', 'v5']

list_short_descriptions()[source]#

A list of unique short descriptions of the downloads.

The unique short descriptions in the given column

Returns:: A list of unique short descriptions extracted from the downloads.
Return type:: list of str

Examples

>>> downloads = [
...     {"alias": "example.txt", "short_description": "shortdesc1"},
...     {"alias": "boo.txt", "short_description": "shortdesc1"},
...     {"alias": "example2.txt", "short_description": "shortdesc2"},
... ]
>>> MGazine(downloads).list_short_descriptions()
['shortdesc1', 'shortdesc2']

property mgnify_analyses: MGnifyMetadata#

property mgnify_assemblies: MGnifyMetadata#

property mgnify_runs: MGnifyMetadata#

property mgnify_samples: MGnifyMetadata#

property mgnify_studies: MGnifyMetadata#

property obs: ResultsHandler#

obs_metadata(df_engine='pandas', expand_nested_dicts=True, drop_duplicates=False, how='left', coalesce=True, for_runs=None, index_col_name='_mgnipy_runs_accs')#

Parameters:

df_engine (Literal ['polars', 'pandas'])
expand_nested_dicts (bool )
drop_duplicates (bool )
coalesce (bool )
for_runs (list [str ] | None)
index_col_name (str )

Return type:

DataFrame | DataFrame

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 short_desc: str #

The short description of the MGazine.

This property returns the FIRST short description of the MGazine, which is derived from the downloads. If multiple short descriptions are present, a warning is issued.

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

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

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#

tsv: handled by stream_pandas() (pandas) or stream_polars() (polars). Gzipped TSVs are supported via the gzip/compression options.
csv: handled by stream_pandas() / stream_polars() (sep=”,”).
txt: handled by stream_txt() (returns full text or yields line chunks).
html: handled by stream_html() (opens URL in browser).
fasta: handled by stream_fasta() (scikit-bio generator).
gff: handled by stream_gff() (scikit-bio generator).
biom: handled by stream_biom() (scikit-bio generator).
gzipped HTTP resources: use stream_gzipped() for a file-like object, or stream_json() for gzipped JSON content.
jsonl / ndjson: handled by stream_jsonl() (pandas or polars modes).
json: handled by stream_json() (returns full JSON or streams via ijson).
tree/newick: handled by stream_tree() (scikit-bio newick reader).
other: if the URL ends with .json it’s streamed via stream_json(); otherwise use the download helper for unsupported binary formats.

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)#

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)#

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)#

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)#

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:

url (str )
chunksize (int | None)
decode (bool )
encoding (str )
errors (str )

Return type:

bytes | str | BufferedReader | TextIOWrapper

stream_html(url, **web_kwargs)#

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)#

Parameters:

url (str )
chunksize (int | None)

Return type:

dict | Generator

stream_jsonl(url, orient=None, chunksize=None, df_engine='pandas', **df_kwargs)#

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)#

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)#

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)#

Parameters:: url (str )
Return type:: Generator

stream_txt(url, chunksize=None, **httpx_kwargs)#

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

to_pandas(**pd_kwargs)[source]#

Return type:: DataFrame

to_polars()[source]#

Return type:: DataFrame

property url_dict: dict [str , dict ]#

Return mapping of alias to URL for all downloads.

Returns:: Dictionary mapping alias -> url (or None when no url is available).
Return type:: dict

Examples

>>> downloads = [{"alias": "example.txt", "url": "http://ex/x"}]
>>> MGazine(downloads).url_dict
{'example.txt': 'http://ex/x'}

property url_list#

Return a list of all download URLs.

Examples

>>> downloads = [{"alias": "example.txt", "url": "http://ex/x"}]
>>> MGazine(downloads).urls
['http://ex/x']

property urls: list [str | None ]#

Return a list of all download URLs. Same as url_list().

Examples

>>> downloads = [{"alias": "example.txt", "url": "http://ex/x"}]
>>> MGazine(downloads).urls
['http://ex/x']

class MGnetizer(resource, all_ids, config=None, client=None, mgnify_metadata=None, detail_proxy=None)[source]#

Bases: CheckpointMixin, ClientManagerMixin

Fetch detailed metadata for a given list of MGnify accessions.

MGnetizer is designed to retrieve the rich metadata from `MGnify`_ for a list of accessions/ids.

Unlike MGnifier or the mgnipy.V2.proxies module, which are designed to search for MGnify lists OR fetch detailed metadata for a single accession at a time, MGnetizer allows for batch processing of multiple accessions. It uses the MGnifyDetail proxy to fetch detailed metadata for each accession and stores the results in a MGnifyMetadata instance.

Parameters:

resource (DetailResourceStr) – The type of resource to fetch metadata for. Must be one of the supported MGnifyDetail resource types (e.g., “study”, “sample”, “run”, etc.).
all_ids (list of str ) – A list of MGnify accessions/ids for which to fetch detailed metadata.
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.
mgnify_metadata (MGnifyMetadata, optional) – An optional MGnifyMetadata instance to store the enriched metadata.
detail_proxy (MGnifyDetail, optional) – An optional MGnifyDetail proxy instance to use for fetching detailed metadata. If not provided, the appropriate proxy will be selected based on the specified resource.

resource#

The type of resource being processed.

Type:: DetailResourceStr

all_ids#

The complete list of MGnify accessions/ids provided during initialization.

Type:: list of str

mgnify_metadata#

The enriched metadata as a MGnifyMetadata instance.

Type:: MGnifyMetadata

metadata#

Alias for mgnify_metadata.

Type:: MGnifyMetadata

params#

A dictionary of parameters used for checkpointing, including the resource type and a sorted list of accessions/ids. Used for caching and resuming enrichment processes.

Type:: dict

async aclose()#

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

Async version of enrich().

See enrich() for details on parameters and behavior.

Parameters:

limit (int | None)
hide_progress (bool )

Return type:

None

property all_ids: list [str ]#: The list of MGnify accessions/ids 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:

request_num (int )
items (Any )

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()#

property datasets: MGazine#: Returns a MGazine instance containing the enriched datasets.

property downloads: list [str ]#: Returns the list of downloads from the enriched metadata.

enrich(limit=200, hide_progress=False)[source]#

Fetches MGnify metadata for the given accessions.

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

Parameters:

limit (Optional[int ], default=200) – An optional integer to limit the number of detaile metadata to retrieve. If set to None, there will be no limit on the number of accessions enriched.
hide_progress (bool , default=False) – Whether to hide the progress bar during enrichment.

Return type:

None

explain()[source]#

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#: Returns the enriched metadata as an MGnifyMetadata instance.

property mgnify_metadata: MGnifyMetadata#: The enriched metadata as an 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:

request_num (int )
items (Any )
include_manifest (bool )

Return type:

None

class MGnifier(resource, *, config=None, params=None, client=None, resolve_auth=True, interactive_auth=False, semaphore=None, **param_kwargs)[source]#

Bases: QuerySet, CheckpointMixin, ClientManagerMixin

MGnifier is a class that provides an interface for querying the MGnify API. It allows users to specify a resource and query parameters, and then fetch results in a paginated manner. The class also includes methods for fetching specific pages, performing bulk fetches, and planning API calls with a dry run.

Parameters:

resource (str ) – The MGnify resource to query (e.g., “studies”, “samples”).
config (MGnipyConfig or dict , optional) – Configuration for MGnipy, either as an MGnipyConfig instance or a dictionary of configuration parameters (default is None).
params (dict , optional) – Query filter parameters (default is None).
client (Client or AuthenticatedClient, optional) – An optional MGnify API client instance to use for requests (default is None).
resolve_auth (bool , optional) – Whether to resolve authentication using the provided config (default is True).
interactive_auth (bool , optional) – Whether to prompt for authentication interactively if needed (default is False).
**param_kwargs – Additional parameters treated as query filters.
semaphore (Optional[asyncio.Semaphore ])

TODO#

async aclose()#

async aget()[source]#

Async alternative to fetch the next page.

Return type:: The next page dict or None when iteration is complete.

Example

mg = MGnifier(“studies”) # doctest: +SKIP next_page = await mg.aget() # doctest: +SKIP

async aget_all(limit=200, *, pages=None, hide_progress=False)[source]#

Asynchronously collect metadata for all (or selected) pages and store results to self.results.

Parameters:

limit (int , optional) – Maximum number of pages to retrieve. If None, retrieves all pages (default is 200).
pages (list of int , optional) – List of page numbers to retrieve. If None, retrieves all pages.
hide_progress (bool , optional) – Whether to hide the progress bar during retrieval (default is False).

async aload_cache()#

Async wrapper for load_cache.

Return type:: list [int ]

async apage(page_num)[source]#

Asynchronously fetch a specific page or range of pages.

Parameters:

page_num (int ) – The page number to retrieve (1-based index).
client (Client, optional) – An optional MGnify API client instance to use for the request. If None, a new client will be initialized.

Returns:

The requested page(s) of results.

Return type:

dict

Examples

mg = MGnifier(“studies”) # doctest: +SKIP page_data = asyncio.run(mg.apage(1)) # doctest: +SKIP

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:

request_num (int )
items (Any )

Return type:

None

property base_url: str #: The base URL for the API, derived from the configuration.

build_queries(**httpx_kwargs)#

Generate a list of query parameter dictionaries for each API request that would be made based on the current parameters. This allows the user to see the specific query parameters for each request before executing them.

Returns:: A list of dictionaries, each containing the query parameters for a corresponding API request.
Return type:: list of dict

property cache_dir: Path | 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()#

property count: int | None #

describe_endpoint(**kwargs)[source]#

Retrieve documentation about the endpoint.

Returns:: Endpoint documentation, or None if unavailable.
Return type:: dict [str , str ] or None

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies")
>>> docs = query.describe_endpoint()

describe_relationships()[source]#

Describe the related resources and their relationships.

Return type:: None

Note

This method is not yet implemented.

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies")
>>> query.describe_relationships()

dry_run()[source]#

Plan the API call by validating parameters and estimating the number of pages and records available. Prints the plan details for the user to review before executing the full data retrieval. This method can be called before get() to ensure that the parameters are valid and to understand the scope of the data retrieval.

Return type:: None

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies", params={"search": "gut"})
>>> query.dry_run()

property endpoint_module: ModuleType #

explain(head=None)[source]#

Print example API URLs that would be called.

Parameters:: head (int , optional) – Maximum number of URLs to print. If None, prints all.
Return type:: None

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies")
>>> query.explain(head=3)

filter(**filters)#

Update the parameters for the API call to filter results.

Parameters:: **filters – Keyword arguments corresponding to the supported parameters for the current resource. These will be used to filter the results returned by the API.
Returns:: A new QuerySet instance with updated parameters for filtering results.
Return type:: QuerySet

first()[source]#

Get the first record from the query results.

Executes the query and returns the first metadata record.

Returns:: The first record as a dictionary, or None if unavailable.
Return type:: dict or None

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies")
>>> first_record = query.first()

get()[source]#

Alternative to getting the next page of results.

Return type:: The next page dict or None when iteration is complete.

Example

mg = MGnifier(“studies”) # doctest: +SKIP next_page = mg.get() # doctest: +SKIP

get_all(limit=200, *, pages=None, hide_progress=False)[source]#

Collect metadata for all (or selected) pages and store results to self.results.

Parameters:

limit (int , optional) – Maximum number of pages to retrieve. If None, retrieves all pages (default is 200).
pages (list of int , optional) – List of page numbers to retrieve. If None, retrieves all pages.
hide_progress (bool , optional) – Whether to hide the progress bar during retrieval (default is False).

property httpx_client: Client#

Get the synchronous httpx client instance from the AuthenticatedClient.

Returns:: The synchronous httpx client instance.
Return type:: httpx.Client

list_relationships()[source]#

Get the names of related resources available from this resource.

Returns:: Names of related resource types (e.g., [“samples”, “analyses”]).
Return type:: list [str ]

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies")
>>> relationships = query.list_relationships()

list_supported_params()[source]#

Get the valid query filter parameters for this resource.

Returns:: Supported parameter names.
Return type:: list [str ]

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies")
>>> params = query.list_supported_params()

list_urls()#

Generate and return a list of URLs for all the API requests that would be made to retrieve the data based on the current parameters. This allows the user to see exactly which endpoints and query parameters will be used in the API calls before executing them.

Returns:: A list of URLs corresponding to each API request that would be made.
Return type:: list of str

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 num_requests: int | None #

page(page_num)[source]#

Retrieve a specific page of metadata for the current resource and parameters. This method allows the user to retrieve metadata one page at a time, which can be useful for previewing data or for manual pagination control.

Parameters:

page_num (int ) – The page number to retrieve (1-based index).
client (Client, optional) – An optional MGnify API client instance to use for the request. If None, a new client will be initialized.

Returns:

A dictionary containing the metadata from the specified page of results, or None if the page is not found.

Return type:

Optional[dict [int , list [dict ]]]

Examples

mg = MGnifier(“studies”) # doctest: +SKIP page_data = mg.page(1) # doctest: +SKIP

property params: dict [str , Any ]#: Get the current parameters for the API request. These parameters are used to filter results and construct the request URL.

preview()[source]#

Get a DataFrame preview of the first page of results.

Quickly check the structure and content of the data without retrieving all pages.

Returns:: DataFrame containing the first page of metadata.
Return type:: pd.DataFrame

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies")
>>> df = query.preview()

property progress: None #: Display the progress of the current query set.

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 request_url: str #

Get the request URL to API based on the current resource and parameters.
This is a single URL that represents the request for the current page of results.

Returns:: The constructed URL for the API request.
Return type:: str

reset_iterator()[source]#: Reset the iterator to start from the beginning.

property resource: SupportedEndpoints#: The type of resource being queried, represented as an instance of SupportedEndpoints.

property results: dict [int , list [dict ]]#: Get the retrieved metadata results, if available. Results are stored in a dictionary with request number (e.g. page number) as keys.

property search_results: MGnifyMetadata#

Get the retrieved metadata results, if available.

Returns:: An object containing the retrieved metadata results and related methods.
Return type:: MGnifyMetadata

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:

request_num (int )
items (Any )
include_manifest (bool )

Return type:

None

class MGnipy(config=None, interactive_auth=False, **config_kwargs)[source]#

Bases: ClientManagerMixin

MGnipy is a Python client for interacting with the MGnify API (https://www.ebi.ac.uk/metagenomics/api/v2/ ).

Provides methods to access different resources (e.g., studies, samples, analyses) and their details, as well as utility methods for listing resources and describing endpoints.

Parameters:

config (MGnipyConfig or dict , optional) – Configuration for MGnipy, either as an MGnipyConfig instance or a dictionary of configuration parameters (default is None).
interactive_auth (bool , optional) – Whether to prompt for authentication interactively if needed (default is False).
**config_kwargs – Additional keyword arguments to pass to the MGnipyConfig constructor if config is not provided. For example, cache_dir can be specified as a keyword argument. e.g. MGnipy(cache_dir=”/path/to/cache”)

Examples

>>> MG = MGnipy(cache_dir=None)  # or MGnipy(cache_dir="/path/to/cache")
>>> MG.cache_dir

async aclose()#

property async_httpx_client: AsyncClient#

Get the asynchronous httpx client instance from the AuthenticatedClient.

Returns:: The asynchronous httpx client instance.
Return type:: httpx.AsyncClient

clear_subcaches()[source]#

Clear the cache for a specific resource or all resources.

Parameters:: resource (str , optional) – The name of the resource to clear the cache for. If None, clears the cache for all resources (default is None).
Return type:: None

close()#

describe_resource(resource, as_dict=False)[source]#

Provides a description of the endpoint from the openapi documentation i.e., https://www.ebi.ac.uk/metagenomics/api/v2/openapi.json

Parameters:

resource (str ) – The name of the resource to describe.
as_dict (bool , optional) – Whether to return the description as a dictionary mapping parameter names to their descriptions (default is False).

Returns:

A dictionary mapping parameter names to their descriptions if as_dict is True, otherwise None.

Return type:

dict of str to str or None

describe_resources(resource=None, as_dict=False)[source]#

Provides a description of the endpoint from the openapi documentation i.e., https://www.ebi.ac.uk/metagenomics/api/v2/openapi.json

Parameters:

resource (str , optional) – The name of the resource to describe.
as_dict (bool , optional) – Whether to return the description as a dictionary mapping parameter names to their descriptions (default is False).

Returns:

A dictionary mapping parameter names to their descriptions if as_dict is True, otherwise None.

Return type:

dict of str to str or None

property httpx_client: Client#

Get the synchronous httpx client instance from the AuthenticatedClient.

Returns:: The synchronous httpx client instance.
Return type:: httpx.Client

list_resources()[source]#: List all supported resources (endpoints) from MGnify API that are supported by mgnipy.

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.

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

class MGnipyConfig(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_prefix_target=None, _env_file=PosixPath('.'), _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, _build_sources=None, *, api_version=SupportedApiVersions.V2, base_url='https://www.ebi.ac.uk/', mg_user=None, mg_password=None, auth_token=None, cache_dir=<factory>)[source]#

Bases: BaseMGnipyConfig

Manage authentication credentials and tokens.

Extension of BaseMGnipyConfig with methods for handling authentication, including obtaining, verifying, and refreshing tokens.

If cache_dir is not set, the auth_token will be cached to working_dir

Roughly the order of events: 1. Check for cached sliding token (comprises access (shorter-lived) and refresh (longer) tokens). 2. If cached token exists, verify the access token’s validity. 3. If valid, move on. 4. If not valid token, try to refresh: get new access token if refresh token is still valid. 5. If that fails, obtain a new one using username/password. 6. Cache the new token for future use.

Parameters:

_case_sensitive (bool | None)
_nested_model_default_partial_update (bool | None)
_env_prefix (str | None)
_env_prefix_target (EnvPrefixTarget | None)
_env_file (DotenvType | None)
_env_file_encoding (str | None)
_env_ignore_empty (bool | None)
_env_nested_delimiter (str | None)
_env_nested_max_split (int | None)
_env_parse_none_str (str | None)
_env_parse_enums (bool | None)
_cli_prog_name (str | None)
_cli_parse_args (bool | list [str ] | tuple [str , ...] | None)
_cli_settings_source (CliSettingsSource[Any] | None)
_cli_parse_none_str (str | None)
_cli_hide_none_type (bool | None)
_cli_avoid_json (bool | None)
_cli_enforce_required (bool | None)
_cli_use_class_docs_for_groups (bool | None)
_cli_exit_on_error (bool | None)
_cli_prefix (str | None)
_cli_flag_prefix_char (str | None)
_cli_implicit_flags (bool | Literal['dual', 'toggle'] | None)
_cli_ignore_unknown_args (bool | None)
_cli_kebab_case (bool | Literal['all', 'no_enums'] | None)
_cli_shortcuts (Mapping[str , str | list [str ]] | None)
_secrets_dir (PathType | None)
_build_sources (tuple [tuple [PydanticBaseSettingsSource, ...], dict [str , Any]] | None)
api_version (SupportedApiVersions)
base_url (HttpUrl)
mg_user (str | None)
mg_password (str | None)
auth_token (str | None)
cache_dir (Path | None)

api_version#

auth_token#

base_url#

cache_dir#

classmethod construct(_fields_set=None, **values)#

Parameters:

_fields_set (set [str ] | None)
values (Any )

Return type:

Self

copy(*, include=None, exclude=None, update=None, deep=False)#

Returns a copy of the model.

!!! warning “Deprecated”: This method is now deprecated; use model_copy instead.

If you need include or exclude, use:

`python {test="skip" lint="skip"} data = self.model_dump(include=include, exclude=exclude, round_trip=True) data = {**data, **(update or {})} copied = self.model_validate(data) `

Parameters:

include (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to include in the copied model.
exclude (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to exclude in the copied model.
update (Dict[str , Any] | None) – Optional dictionary of field-value pairs to override field values in the copied model.
deep (bool ) – If True, the values of fields that are Pydantic models will be deep-copied.

Returns:

A copy of the model with included, excluded and updated fields as specified.

Return type:

Self

dict(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False)#

Parameters:

include (set [int ] | set [str ] | Mapping [int , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | Mapping [str , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | None)
exclude (set [int ] | set [str ] | Mapping [int , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | Mapping [str , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | None)
by_alias (bool )
exclude_unset (bool )
exclude_defaults (bool )
exclude_none (bool )

Return type:

Dict [str , Any ]

classmethod from_orm(obj)#

Parameters:: obj (Any )
Return type:: Self

json(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, encoder=PydanticUndefined, models_as_dict=PydanticUndefined, **dumps_kwargs)#

Parameters:

include (set [int ] | set [str ] | Mapping [int , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | Mapping [str , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | None)
exclude (set [int ] | set [str ] | Mapping [int , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | Mapping [str , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | None)
by_alias (bool )
exclude_unset (bool )
exclude_defaults (bool )
exclude_none (bool )
encoder (Callable [[Any ], Any ] | None)
models_as_dict (bool )
dumps_kwargs (Any )

Return type:

str

mg_password#

mg_user#

model_computed_fields = {}#

model_config = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': '', 'env_prefix_target': 'variable', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod model_construct(_fields_set=None, **values)#

Creates a new instance of the Model class with validated data.

Creates a new model setting __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed.

!!! note: model_construct() generally respects the model_config.extra setting on the provided model. That is, if model_config.extra == ‘allow’, then all extra passed values are added to the model instance’s __dict__ and __pydantic_extra__ fields. If model_config.extra == ‘ignore’ (the default), then all extra passed values are ignored. Because no validation is performed with a call to model_construct(), having model_config.extra == ‘forbid’ does not result in an error if extra values are passed, but they will be ignored.

Parameters:

_fields_set (set [str ] | None) – A set of field names that were originally explicitly set during instantiation. If provided, this is directly used for the [model_fields_set][pydantic.BaseModel.model_fields_set] attribute. Otherwise, the field names from the values argument will be used.
values (Any ) – Trusted or pre-validated data dictionary.

Returns:

A new instance of the Model class with validated data.

Return type:

Self

model_copy(*, update=None, deep=False)#

!!! abstract “Usage Documentation”: [model_copy](../concepts/models.md#model-copy)

Returns a copy of the model.

!!! note: The underlying instance’s [__dict__][object.__dict__] attribute is copied. This might have unexpected side effects if you store anything in it, on top of the model fields (e.g. the value of [cached properties][functools.cached_property]).

Parameters:

update (Mapping [str , Any ] | None) – Values to change/add in the new model. Note: the data is not validated before creating the new model. You should trust this data.
deep (bool ) – Set to True to make a deep copy of the model.

Returns:

New model instance.

Return type:

Self

model_dump(*, mode='python', include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, exclude_computed_fields=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)#

!!! abstract “Usage Documentation”: [model_dump](../concepts/serialization.md#python-mode)

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Parameters:

mode (Literal ['json', 'python'] | str) – The mode in which to_python should run. If mode is ‘json’, the output will only contain JSON serializable types. If mode is ‘python’, the output may contain non-JSON-serializable Python objects.
include (set [int ] | set [str ] | Mapping [int , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | Mapping [str , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | None) – A set of fields to include in the output.
exclude (set [int ] | set [str ] | Mapping [int , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | Mapping [str , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | None) – A set of fields to exclude from the output.
context (Any | None) – Additional context to pass to the serializer.
by_alias (bool | None) – Whether to use the field’s alias in the dictionary key if defined.
exclude_unset (bool ) – Whether to exclude fields that have not been explicitly set.
exclude_defaults (bool ) – Whether to exclude fields that are set to their default value.
exclude_none (bool ) – Whether to exclude fields that have a value of None.
exclude_computed_fields (bool ) – Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.
round_trip (bool ) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].
warnings (bool | Literal ['none', 'warn', 'error']) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].
fallback (Callable [[Any ], Any ] | None) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.
serialize_as_any (bool ) – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A dictionary representation of the model.

Return type:

dict [str , Any ]

model_dump_json(*, indent=None, ensure_ascii=False, include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, exclude_computed_fields=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)#

!!! abstract “Usage Documentation”: [model_dump_json](../concepts/serialization.md#json-mode)

Generates a JSON representation of the model using Pydantic’s to_json method.

Parameters:

indent (int | None) – Indentation to use in the JSON output. If None is passed, the output will be compact.
ensure_ascii (bool ) – If True, the output is guaranteed to have all incoming non-ASCII characters escaped. If False (the default), these characters will be output as-is.
include (set [int ] | set [str ] | Mapping [int , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | Mapping [str , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | None) – Field(s) to include in the JSON output.
exclude (set [int ] | set [str ] | Mapping [int , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | Mapping [str , set [int ] | set [str ] | Mapping [int , IncEx | bool ] | Mapping [str , IncEx | bool ] | bool ] | None) – Field(s) to exclude from the JSON output.
context (Any | None) – Additional context to pass to the serializer.
by_alias (bool | None) – Whether to serialize using field aliases.
exclude_unset (bool ) – Whether to exclude fields that have not been explicitly set.
exclude_defaults (bool ) – Whether to exclude fields that are set to their default value.
exclude_none (bool ) – Whether to exclude fields that have a value of None.
exclude_computed_fields (bool ) – Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.
round_trip (bool ) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].
warnings (bool | Literal ['none', 'warn', 'error']) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].
fallback (Callable [[Any ], Any ] | None) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.
serialize_as_any (bool ) – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A JSON string representation of the model.

Return type:

str

property model_extra: dict [str , Any ] | None #

Get extra fields set during validation.

Returns:: A dictionary of extra fields, or None if config.extra is not set to “allow”.

model_fields = {'api_version': FieldInfo(annotation=SupportedApiVersions, required=False, default=<SupportedApiVersions.V2: 'v2'>, description="API version to use. Supported values are 'v2', and 'latest'."), 'auth_token': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Authentication token for API access. If provided, it will be used for authenticated requests.', repr=False), 'base_url': FieldInfo(annotation=HttpUrl, required=False, default='https://www.ebi.ac.uk/', description='Base URL for the MGnify API', validate_default=True), 'cache_dir': FieldInfo(annotation=Union[Path, NoneType], required=False, default_factory=<lambda>, description=('Cache directory for storing API responses or other temp things. Defaults to a platform-appropriate cache dir via `platformdirs`. Set to None to disable disk caching.',)), 'mg_password': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Password for basic authentication (if required)', repr=False), 'mg_user': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Username for basic authentication (if required)', repr=False)}#

property model_fields_set: set [str ]#

Returns the set of fields that have been explicitly set on this model instance.

Returns:

A set of strings representing the fields that have been set,: i.e. that were not filled from defaults.

classmethod model_json_schema(by_alias=True, ref_template='#/$defs/{model}', schema_generator=<class 'pydantic.json_schema.GenerateJsonSchema'>, mode='validation', *, union_format='any_of')#

Generates a JSON schema for a model class.

Parameters:

by_alias (bool ) – Whether to use attribute aliases or not.
ref_template (str ) – The reference template.
union_format (Literal ['any_of', 'primitive_type_array']) –
The format to use when combining schemas from unions together. Can be one of:
- ’any_of’: Use the [anyOf](https://json-schema.org/understanding-json-schema/reference/combining#anyOf )
keyword to combine schemas (the default). - ‘primitive_type_array’: Use the [type](https://json-schema.org/understanding-json-schema/reference/type ) keyword as an array of strings, containing each type of the combination. If any of the schemas is not a primitive type (string, boolean, null, integer or number) or contains constraints/metadata, falls back to any_of.
schema_generator (type [GenerateJsonSchema]) – To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications
mode (Literal ['validation', 'serialization']) – The mode in which to generate the schema.

Returns:

The JSON schema for the given model class.

Return type:

dict [str , Any ]

classmethod model_parametrized_name(params)#

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic BaseModels.

Parameters:: params (tuple [type [Any ], ...]) – Tuple of types of the class. Given a generic class Model with 2 type variables and a concrete model Model[str, int], the value (str, int) would be passed to params.
Returns:: String representing the new class where params are passed to cls as type variables.
Raises:: TypeError – Raised when trying to generate concrete names for non-generic models.
Return type:: str

model_post_init(context, /)#

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

Parameters:: context (Any )
Return type:: None

classmethod model_rebuild(*, force=False, raise_errors=True, _parent_namespace_depth=2, _types_namespace=None)#

Try to rebuild the pydantic-core schema for the model.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

force (bool ) – Whether to force the rebuilding of the model schema, defaults to False.
raise_errors (bool ) – Whether to raise errors, defaults to True.
_parent_namespace_depth (int ) – The depth level of the parent namespace, defaults to 2.
_types_namespace (MappingNamespace | None) – The types namespace, defaults to None.

Returns:

Returns None if the schema is already “complete” and rebuilding was not required. If rebuilding _was_ required, returns True if rebuilding was successful, otherwise False.

Return type:

bool | None

classmethod model_validate(obj, *, strict=None, extra=None, from_attributes=None, context=None, by_alias=None, by_name=None)#

Validate a pydantic model instance.

Parameters:

obj (Any ) – The object to validate.
strict (bool | None) – Whether to enforce types strictly.
extra (Literal ['allow', 'ignore', 'forbid'] | None) – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.
from_attributes (bool | None) – Whether to extract data from object attributes.
context (Any | None) – Additional context to pass to the validator.
by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.
by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Raises:

ValidationError – If the object could not be validated.

Returns:

The validated model instance.

Return type:

Self

classmethod model_validate_json(json_data, *, strict=None, extra=None, context=None, by_alias=None, by_name=None)#

!!! abstract “Usage Documentation”: [JSON Parsing](../concepts/json.md#json-parsing)

Validate the given JSON data against the Pydantic model.

Parameters:

json_data (str | bytes | bytearray ) – The JSON data to validate.
strict (bool | None) – Whether to enforce types strictly.
extra (Literal ['allow', 'ignore', 'forbid'] | None) – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.
context (Any | None) – Extra variables to pass to the validator.
by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.
by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Raises:

ValidationError – If json_data is not a JSON string or the object could not be validated.

Return type:

Self

classmethod model_validate_strings(obj, *, strict=None, extra=None, context=None, by_alias=None, by_name=None)#

Validate the given object with string data against the Pydantic model.

Parameters:

obj (Any ) – The object containing string data to validate.
strict (bool | None) – Whether to enforce types strictly.
extra (Literal ['allow', 'ignore', 'forbid'] | None) – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.
context (Any | None) – Extra variables to pass to the validator.
by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.
by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Return type:

Self

obtain_auth_token(*, username=None, password=None)[source]#

Obtains an authentication token using the MGnify username and password. If credentials are not available, can prompt the user to enter them.

Parameters:

username (str , optional) – MGnify username. If not provided, will attempt to use the configured username or prompt the user.
password (str , optional) – MGnify password. If not provided, will attempt to use the configured password or prompt the user.

Returns:

The obtained authentication token, or None if the token could not be obtained.

Return type:

str or None

Example

config = MGnipyConfig(mg_user=”myuser”, mg_password=”mypassword”) token = config.obtain_auth_token() # doctest: +SKIP

classmethod parse_file(path, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)#

Parameters:

path (str | Path)
content_type (str | None)
encoding (str )
proto (DeprecatedParseProtocol | None)
allow_pickle (bool )

Return type:

Self

classmethod parse_obj(obj)#

Parameters:: obj (Any )
Return type:: Self

classmethod parse_raw(b, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)#

Parameters:

b (str | bytes )
content_type (str | None)
encoding (str )
proto (DeprecatedParseProtocol | None)
allow_pickle (bool )

Return type:

Self

refresh_auth_token(token=None)[source]#

Refresh the provided authentication token using the sliding token refresh endpoint. If no token is provided, it will attempt to refresh the token stored in the config.

Parameters:: token (str | None)
Return type:: str | None

resolve_auth_token(*, interactive=False)[source]#

Resolve a valid authentication token by checking the current token, verifying it, and refreshing or obtaining a new one as needed.

Parameters:: interactive (bool , optional) – If True, prompts the user to input credentials if they are not found in the config. Default is True.
Return type:: None

Example

config = MGnipyConfig(mg_user=”myuser”, mg_password=”mypassword”) config.resolve_auth_token()

classmethod schema(by_alias=True, ref_template='#/$defs/{model}')#

Parameters:

by_alias (bool )
ref_template (str )

Return type:

Dict [str , Any ]

classmethod schema_json(*, by_alias=True, ref_template='#/$defs/{model}', **dumps_kwargs)#

Parameters:

by_alias (bool )
ref_template (str )
dumps_kwargs (Any )

Return type:

str

serialize_base_url(v)#

Custom serializer for the base_url field to ensure it is always represented as a string.

Parameters:: v (HttpUrl)
Return type:: str

classmethod settings_customise_sources(settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings)#

Define the sources and their order for loading the settings values.

Parameters:

settings_cls (type [BaseSettings]) – The Settings class.
init_settings (PydanticBaseSettingsSource) – The InitSettingsSource instance.
env_settings (PydanticBaseSettingsSource) – The EnvSettingsSource instance.
dotenv_settings (PydanticBaseSettingsSource) – The DotEnvSettingsSource instance.
file_secret_settings (PydanticBaseSettingsSource) – The SecretsSettingsSource instance.

Returns:

A tuple containing the sources and their order for loading the settings values.

Return type:

tuple [PydanticBaseSettingsSource, …]

classmethod update_forward_refs(**localns)#

Parameters:: localns (Any )
Return type:: None

classmethod validate(value)#

Parameters:: value (Any )
Return type:: Self

verify_auth_token(token=None)[source]#

Verify the validity of the provided authentication token Makes request to the token verification endpoint If no token is provided, it will attempt to verify the token stored in the config.

Parameters:: token (str , optional) – The authentication token to verify. If not provided, the method will use the token stored in the config.
Returns:: True if the token is valid, False otherwise.
Return type:: bool

Example

config = MGnipyConfig() is_valid = config.verify_auth_token(“your_auth_token”) # doctest: +SKIP

class MTG(dataset, *, var_cols=None, var_index=None, obs_index='_mgnipy_runs_accs', mgnify_studies=None, mgnify_analyses=None, mgnify_runs=None, mgnify_samples=None, mgnify_assemblies=None, biosamples_metadata=None, obs=None)[source]#

Bases: MetadataSettersMixin

MGic the Gatherer combines a MGnify dataset with its metadata.

The MGic gatherer (MTG) takes a dataset as pandas or polars dataframe and MGnify or BioSamples metadata and combines them into a single object. MTG can be used to enrich the dataset with metadata, and to convert the dataset into different formats such as pandas, polars, or anndata.

Parameters:

dataset (pandas.DataFrame or polars.DataFrame) – The dataset to be combined with metadata. This can be a pandas or polars dataframe.
var_cols (list of str , optional) – A list of column names in the dataset that are considered variable columns. These columns will be in var_metadata() and excluded from obs_metadata()
mgnify_[studies|analyses|runs|samples|assemblies] (list of dict , optional) – Lists of dictionaries containing metadata for each respective MGnify dataset.
biosamples_metadata (list of dict , optional) – A list of dictionaries containing metadata for BioSamples.
var_index (str | None)
obs_index (str )
mgnify_studies (list [dict [str , Any]] | None)
mgnify_analyses (list [dict [str , Any]] | None)
mgnify_runs (list [dict [str , Any]] | None)
mgnify_samples (list [dict [str , Any]] | None)
mgnify_assemblies (list [dict [str , Any]] | None)
obs (list [dict [str , Any]] | None)

runs_accessions#

A list of all run accessions in the dataset. This is derived from the columns of the dataset that are not in var_cols.

Type:: list

X(df_engine='pandas')[source]#

Gets the feature matrix (X) from the dataset.

Basically transposes.

Parameters:: df_engine (Literal["polars", "pandas"], optional) – The DataFrame engine to use for the output. If “polars” is specified, a polars.DataFrame is returned; if “pandas” is specified, a pandas.DataFrame is returned.
Returns:: The feature matrix (X) containing the non-var columns from the dataset
Return type:: pl.DataFrame or pd.DataFrame

append_biosamples_metadata(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_analyses(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_assemblies(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_runs(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_samples(value)#

Parameters:: value (dict [str , Any ])

append_mgnify_studies(value)#

Parameters:: value (dict [str , Any ])

append_obs(value)#

Parameters:: value (dict [str , Any ])

property available_metadata_sets: list [str ]#

Return a list of available metadata sets in the MTG.

This property checks which metadata sets (e.g., studies, analyses, runs, samples, assemblies, biosamples) are non-empty and returns their names as a list.

Returns:: A list of names of non-empty metadata sets available in the MTG.
Return type:: list of str

property biosamples_metadata: ResultsHandler#

property mgnify_analyses: MGnifyMetadata#

property mgnify_assemblies: MGnifyMetadata#

property mgnify_runs: MGnifyMetadata#

property mgnify_samples: MGnifyMetadata#

property mgnify_studies: MGnifyMetadata#

property obs: ResultsHandler#

obs_metadata(*args, **kwargs)[source]#

Return type:: DataFrame | DataFrame

property runs_accessions: list #

to_anndata(drop_duplicates=True, **anndata_kwargs)[source]#

Parameters:: drop_duplicates (bool )
Return type:: AnnData

to_pandas()[source]#

Return type:: DataFrame

to_polars()[source]#

Return type:: DataFrame

var_metadata(df_engine='pandas')[source]#

Return the variable metadata as a dataframe.

Parameters:: df_engine (str , optional) – The dataframe engine to use. Can be “polars” or “pandas”. Default is “pandas”.
Returns:: A dataframe containing the variable metadata.
Return type:: pd.DataFrame or pl.DataFrame

mgnipy package

Contents

mgnipy package#

Supported formats and their handlers#

Subpackages#

Submodules#