mgnipy.V2.mgnifier package

mgnipy.V2.mgnifier package#

MGnifier: A class for querying the MGnify API with support for caching, pagination, and metadata retrieval.

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

config: MGnipyConfig#

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

emgapi_handler: DescribeEmgapiModule#

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

mgnipy.V2.mgnifier package

Contents

mgnipy.V2.mgnifier package#

Submodules#