Source code for mgnipy.mgnipy

from typing import Optional

from mgnipy._models.config import MgnipyConfig
from mgnipy._models.CONSTANTS import SupportedEndpoints
from mgnipy.V2.proxies import (
    V2_ENDPOINT_DETAIL_PROXIES,
    V2_ENDPOINT_LIST_PROXIES,
)

V2_ALL_PROXIES = V2_ENDPOINT_DETAIL_PROXIES | V2_ENDPOINT_LIST_PROXIES


[docs] class MGnipy: """ """ def __init__(self, **config): self._config = MgnipyConfig(**config) self._endpoints = self.list_resources() def __getattr__(self, name: str): endpoint = SupportedEndpoints.validate(name) if endpoint in V2_ENDPOINT_LIST_PROXIES: list_cls = V2_ENDPOINT_LIST_PROXIES[endpoint] return list_cls(config=self._config.model_dump(mode="json")) if endpoint in V2_ENDPOINT_DETAIL_PROXIES: detail_cls = V2_ENDPOINT_DETAIL_PROXIES[endpoint] # Return a callable so required args like accession/biome_lineage # are provided when user calls MG.study(...), MG.biome(...), etc. def _detail_factory(id: Optional[str] = None, **kwargs): return detail_cls( id=id, config=self._config.model_dump(mode="json"), **kwargs ) return _detail_factory raise AttributeError( f"{type(self).__name__} has no endpoint attribute {name!r}" )
[docs] def list_resources(self): return [endpoint.value for endpoint in SupportedEndpoints]
[docs] def describe_resource( self, resource: str, as_dict: bool = False ) -> dict[str, str] | None: """ Describe the supported parameters for a given resource by parsing the docstring of the corresponding endpoint module. 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 ------- dict of str to str or None A dictionary mapping parameter names to their descriptions if as_dict is True, otherwise None. """ try: endpoint = SupportedEndpoints.validate(resource) except ValueError: print( f"Resource '{resource}' is not supported. Supported resources are: {', '.join(self.list_resources())}" ) return None proxy_cls = V2_ALL_PROXIES[endpoint] proxy = proxy_cls(config=self._config.model_dump(mode="json")) return proxy.describe_endpoint(as_dict=as_dict)
[docs] def describe_resources( self, resource: Optional[str] = None, as_dict: bool = False ) -> dict[str, str] | None: if resource is not None: return self.describe_resource(resource, as_dict=as_dict) descriptions = {} for endpoint in SupportedEndpoints: desc = self.describe_resource(endpoint.value, as_dict=as_dict) if desc is not None: descriptions[endpoint.value] = desc return descriptions