mgnipy.V2.proxies.studies module

mgnipy.V2.proxies.studies module#

class PrivateStudies(*, params=None, config=None, **kwargs)[source]#

Bases: MGnifyList

Parameters:

params (Optional[dict [str , Any]])
config (MGnipyConfig)

RESOURCE: ClassVar [Literal ['private_studies']] = 'private_studies'#

async aclose()#

async aenrich_details(limit=200, hide_progress=False)#

Async version of enrich_details that retrieves details for each item in the MGnifyList asynchronously.

Parameters:

limit (Optional[int ], default=200) – An optional integer to limit the number of items to enrich. If not provided, it defaults to 200. If set to None, there will be no limit on the number of items enriched.
hide_progress (bool , default=False) – A boolean flag to control the display of the progress bar. If set to True, the progress bar will be hidden.

Returns:

This method does not return anything. It updates the internal state of the MGnifyList instance by populating the .details .details_df and .metadata.results with the details of each item.

Return type:

None

async aget()#

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

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

Async variant of get_detail.

Returns:: The next detail proxy, or None if no more details to iterate.
Return type:: MGnifyDetail or None

property aiter_details: AsyncIterator [dict ]#

Async version of iter_details.

Returns:: An async iterator that yields MGnifyDetail results one by one, fetched on demand.
Return type:: AsyncIterator[dict ]

async aload_cache()#

Async wrapper for load_cache.

Return type:: list [int ]

async apage(page_num)#

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.

child_resource: str #

clear_cache()#

Remove all cached pages for this set of queries.

Return type:: None

close()#

config: MGnipyConfig#

property count: int | None #

property datasets#: A property that returns an MGazine instance containing the downloads information for the study.

describe_endpoint(**kwargs)#

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

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

property downloads: list [dict [str , Any ]] | None #

Get a list of all download links from the detailed metadata.

Returns:: A list of dictionaries containing download information, or None if no details are available.
Return type:: list [dict [str , Any]] or None

downloads_df(**pd_kwargs)[source]#

Return type:: DataFrame

dry_run()#

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 #

enrich_details(limit=200, hide_progress=False)#

Gets the details for each mgnify list item. Iterates through the accessions/ids (.search_results.ids) and retrieves their details using the corresponding detail proxy (e.g., RunDetail for Runs).

Parameters:

limit (Optional[int ], default=200) – An optional integer to limit the number of runs to enrich. If not provided, it defaults to 200. If set to None, there will be no limit on the number of runs enriched.
hide_progress (bool , default=False) – A boolean flag to control the display of the progress bar. If set to True, the progress bar will be hidden.

Returns:

This method does not return anything. It updates the internal state of the MGnifyList instance by populating the .details .details_df and .metadata.results with the details of each item.

Return type:

None

explain(head=None)#

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

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

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

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

get_detail()#

Get the next MGnifyDetail based on current _detail_index. Updates _last_successful_detail on success.

Returns:: The next detail proxy, or None if no more details to iterate.
Return type:: MGnifyDetail or None

Example

>>> from mgnipy.V2.proxies import Studies
>>> studies = Studies(search="tomato")
>>> studies.get_all()
>>> first_detail = studies.get_detail()
>>> second_detail = studies.get_detail()

property httpx_client: Client#

Get the synchronous httpx client instance from the AuthenticatedClient.

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

property iter_details: Iterator [dict ]#

Yield MGnifyDetail results one by one.

Returns:: An iterator that yields MGnifyDetail results one by one, fetched on demand.
Return type:: Iterator[dict ]

Examples

>>> from mgnipy.V2.proxies import Studies
>>> studies = Studies()
>>> result_dict = next(studies.iter_details)

list_relationships()#

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

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 metadata: MGnifyMetadata#

property mgnify_details: list [MGnifyDetail]#

property num_requests: int | None #

page(page_num)#

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

page_size(n)#

Set the page size for paginated API calls.

Parameters:: n (int )
Returns:: A new MGnifyList instance with the updated page size parameter.
Return type:: MGnifyList

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

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()#: 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 Studies(*, params=None, config=None, **kwargs)[source]#

Bases: MGnifyList

Parameters:

params (Optional[dict [str , Any]])
config (MGnipyConfig)

RESOURCE: ClassVar [Literal ['studies']] = 'studies'#

async aclose()#

async aenrich_details(limit=200, hide_progress=False)#

Async version of enrich_details that retrieves details for each item in the MGnifyList asynchronously.

Parameters:

limit (Optional[int ], default=200) – An optional integer to limit the number of items to enrich. If not provided, it defaults to 200. If set to None, there will be no limit on the number of items enriched.
hide_progress (bool , default=False) – A boolean flag to control the display of the progress bar. If set to True, the progress bar will be hidden.

Returns:

This method does not return anything. It updates the internal state of the MGnifyList instance by populating the .details .details_df and .metadata.results with the details of each item.

Return type:

None

async aget()#

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

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

Async variant of get_detail.

Returns:: The next detail proxy, or None if no more details to iterate.
Return type:: MGnifyDetail or None

property aiter_details: AsyncIterator [dict ]#

Async version of iter_details.

Returns:: An async iterator that yields MGnifyDetail results one by one, fetched on demand.
Return type:: AsyncIterator[dict ]

async aload_cache()#

Async wrapper for load_cache.

Return type:: list [int ]

async apage(page_num)#

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.

child_resource: str #

clear_cache()#

Remove all cached pages for this set of queries.

Return type:: None

close()#

config: MGnipyConfig#

property count: int | None #

property datasets#: A property that returns an MGazine instance containing the downloads information for the study.

describe_endpoint(**kwargs)#

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

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

property downloads: list [dict [str , Any ]] | None #

Get a list of all download links from the detailed metadata.

Returns:: A list of dictionaries containing download information, or None if no details are available.
Return type:: list [dict [str , Any]] or None

downloads_df(**pd_kwargs)[source]#

Return type:: DataFrame

dry_run()#

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 #

enrich_details(limit=200, hide_progress=False)#

Gets the details for each mgnify list item. Iterates through the accessions/ids (.search_results.ids) and retrieves their details using the corresponding detail proxy (e.g., RunDetail for Runs).

Parameters:

limit (Optional[int ], default=200) – An optional integer to limit the number of runs to enrich. If not provided, it defaults to 200. If set to None, there will be no limit on the number of runs enriched.
hide_progress (bool , default=False) – A boolean flag to control the display of the progress bar. If set to True, the progress bar will be hidden.

Returns:

This method does not return anything. It updates the internal state of the MGnifyList instance by populating the .details .details_df and .metadata.results with the details of each item.

Return type:

None

explain(head=None)#

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

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

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

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

get_detail()#

Get the next MGnifyDetail based on current _detail_index. Updates _last_successful_detail on success.

Returns:: The next detail proxy, or None if no more details to iterate.
Return type:: MGnifyDetail or None

Example

>>> from mgnipy.V2.proxies import Studies
>>> studies = Studies(search="tomato")
>>> studies.get_all()
>>> first_detail = studies.get_detail()
>>> second_detail = studies.get_detail()

property httpx_client: Client#

Get the synchronous httpx client instance from the AuthenticatedClient.

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

property iter_details: Iterator [dict ]#

Yield MGnifyDetail results one by one.

Returns:: An iterator that yields MGnifyDetail results one by one, fetched on demand.
Return type:: Iterator[dict ]

Examples

>>> from mgnipy.V2.proxies import Studies
>>> studies = Studies()
>>> result_dict = next(studies.iter_details)

list_relationships()#

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

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 metadata: MGnifyMetadata#

property mgnify_details: list [MGnifyDetail]#

property num_requests: int | None #

page(page_num)#

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

page_size(n)#

Set the page size for paginated API calls.

Parameters:: n (int )
Returns:: A new MGnifyList instance with the updated page size parameter.
Return type:: MGnifyList

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

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()#: 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 StudyDetail(id=None, *, accession=None, config=None, **kwargs)[source]#

Bases: MGnifyDetail

Parameters:

id (Optional[str ])
accession (Optional[str ])
config (MGnipyConfig)

RESOURCE: ClassVar [Literal ['study']] = 'study'#

async aclose()#

async aget()#

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

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 aget_list(resource, *, fetch=True, explain=False)#

Get list proxy for a specific accession/pubmed_id/catalogue_id detail.

Parameters:

resource (str ) – Valid list resource name e.g. in list_relationships(), such as “samples” for a study detail, or “analyses” for a run detail.
fetch (bool ) – Whether to immediately fetch the detail after creating the proxy.
explain (bool ) – Whether to print example URLs that would be called.

Returns:

A proxy for the next resource.

Return type:

MGnifyList

async aload_cache()#

Async wrapper for load_cache.

Return type:: list [int ]

async apage(page_num)#

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 #

property datasets#: A property that returns an MGazine instance containing the downloads information for the study.

describe_endpoint(**kwargs)#

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

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

property downloads: list [dict [str , Any ]]#: Get the list of downloads for this detail.

downloads_df(**pd_kwargs)#

Looking for a “downloads” field in the metadata results and return as a DataFrame if found.

Return type:: DataFrame | None

dry_run()#

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

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

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

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

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

get_list(resource, *, fetch=True, explain=False)#

Get list proxy for a specific accession/pubmed_id/catalogue_id detail.

Parameters:

resource (str ) – Valid child resource name e.g. in list_relationships(), such as “samples” for a study detail, or “analyses” for a run detail.
fetch (bool ) – Whether to immediately fetch the detail after creating the proxy.
explain (bool ) – Whether to print example URLs that would be called.

Returns:

A proxy for the next resource.

Return type:

MGnifyList

property httpx_client: Client#

Get the synchronous httpx client instance from the AuthenticatedClient.

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

property identifier: str | None #

Get the identifier value from the query parameters.

Used for constructing URLs to related resources.

Returns:: The identifier value, or None if not set.
Return type:: str or None

Examples

>>> from mgnipy.V2.mgnifier import MGnifier
>>> query = MGnifier("studies", accession="MGYS000000001", config={})
>>> query.identifier

list_relationships()#

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

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

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

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()#: 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.proxies.studies module

Contents

mgnipy.V2.proxies.studies module#