Source code for mgnipy.V2.mgnifier.describe
from __future__ import annotations
import inspect
import logging
from mgnipy._models.constants.CONSTANTS import SupportedEndpoints
logger = logging.getLogger(__name__)
from copy import deepcopy
from math import ceil
import os
from types import ModuleType
from typing import Any, Optional
from urllib.parse import urlencode
from mgnipy._shared_helpers.httpx_helpers import init_httpx_client
from mgnipy._shared_helpers.parsers import get_docstring, parse_docstring
from mgnipy.V2.mgnifier.endpoints import ALL_LIST_ENDPOINTS, ID_PARAM, PRIVATE_ENDPOINTS
[docs]
class DescribeEmgapiModule:
"""
A class to describe and interact with an endpoint module from the metagenomics API v2.
This class provides methods to list supported parameters, validate keyword arguments, construct URLs, and retrieve endpoint documentation.
It also includes functionality to determine if the endpoint is private or a list endpoint, and to calculate pagination details based on the number of items and page size.
Parameters
----------
endpoint_module : ModuleType
The endpoint module to describe.
"""
def __init__(
self,
endpoint_module: ModuleType,
):
self.endpoint_module = endpoint_module
self.default_page_size: int = 25
[docs]
def list_supported_params(self) -> list[str]:
"""
Lists supported keyword arguments for the endpoint module.
Returns
-------
list of str
List of supported keyword argument names.
"""
sig = inspect.signature(self.endpoint_module._get_kwargs)
return list(sig.parameters.keys())
[docs]
def validate_endpoint_kwargs(self, **kwargs) -> dict[str, Any]:
"""
Validates the provided keyword arguments against the supported parameters of the endpoint module.
Parameters
----------
**kwargs
Keyword arguments to validate.
Returns
-------
dict of str to Any
The validated keyword arguments.
Raises
------
ValueError
If any provided keyword argument is not supported by the endpoint module.
"""
return self.endpoint_module._get_kwargs(**kwargs)
@property
def emgapi_resource(self) -> Optional[str]:
"""
Retrieves the name of the endpoint resource based on the endpoint module.
E.g., if the endpoint module corresponds to the "studies" resource, this property would return "studies".
Returns
-------
str or None
The name of the endpoint resource, or None if the endpoint module is not set.
"""
return os.path.basename(os.path.dirname(self.endpoint_module.__file__))
[docs]
def sub_url(self, **kwargs) -> Optional[str]:
"""
Constructs the sub-URL for the endpoint based on the current parameters.
Returns
-------
str or None
The constructed sub-URL, or None if the endpoint module is not set.
"""
_kwargs = self.validate_endpoint_kwargs(**kwargs)
_end_url: str = _kwargs.get(
"url", f"/metagenomics/api/v2/{self.emgapi_resource}/"
).strip("/")
return _end_url
[docs]
def resolve_query_string(self, **kwargs) -> str:
"""
Resolves the query string for the endpoint based on the current parameters.
Parameters
----------
**kwargs
Keyword arguments to validate and include in the query string.
Returns
-------
str
The resolved query string.
"""
_kwargs = self.validate_endpoint_kwargs(**kwargs)
# get validated params if any
params = _kwargs.get("params", {})
# encode params for url
return urlencode(params, doseq=True)
[docs]
def url_path(self, **kwargs) -> str:
"""
Constructs the full URL path for the endpoint based on the current parameters.
Parameters
----------
**kwargs
Keyword arguments to validate and include in the URL construction.
Returns
-------
str
The constructed URL path.
"""
_end_url = self.sub_url(**kwargs)
query_string = self.resolve_query_string(**kwargs)
return f"{_end_url}?{query_string}" if query_string else _end_url
@property
def emgapi_docs(self) -> str:
return get_docstring(self.endpoint_module, "sync")
[docs]
def describe_endpoint(self, as_dict: bool = False) -> dict[str, str] | None:
"""
Provides a description of the endpoint from the openapi documentation i.e., https://www.ebi.ac.uk/metagenomics/api/v2/openapi.json
Parameters
----------
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.
"""
return parse_docstring(self.emgapi_docs, as_dict=as_dict)
@property
def is_private(self) -> bool:
"""Checks if the endpoint module corresponds to a private only endpoint."""
return self.endpoint_module in PRIVATE_ENDPOINTS
@property
def is_list_endpoint(self) -> bool:
"""Checks if the endpoint module corresponds to a list endpoint."""
return self.endpoint_module in ALL_LIST_ENDPOINTS
[docs]
def get_num_items(self, params: Optional[dict] = None) -> Optional[int]:
_params: dict[str, Any] = deepcopy(params) or {}
if not self.is_list_endpoint:
return 1
# otherwise mini request with small page_size to get count
_params.update({"page_size": 1})
with init_httpx_client() as c:
response = self.endpoint_module.sync_detailed(client=c, **_params)
if response.status_code == 200:
return response.parsed.to_dict().get("count", 0)
else:
logger.warning(
f"Failed to retrieve count for endpoint {self.emgapi_resource}. Status code: {response.status_code}"
)
[docs]
def get_num_pages(
self, num_items: Optional[int], page_size: Optional[int] = None
) -> Optional[int]:
"""Calculates the total number of pages based on the total count and default page size."""
if num_items is None:
return None
ps = page_size or self.default_page_size
return ceil(num_items / ps) or 1
[docs]
def page_param_iter(self, num_pages: int) -> list[dict[str, int]]:
"""Generates a list of parameter dictionaries for each page based on the total number of pages."""
return [{"page": page} for page in range(1, num_pages + 1)]
@property
def id_param_key(self) -> str:
"""Get the parameter name used to identify this resource.
Returns
-------
str
The identifier parameter (e.g., "accession", "biome_lineage").
"""
try:
logger.debug(
f"Attempting to retrieve identifier parameter key for resource '{self.emgapi_resource}'"
)
return ID_PARAM[SupportedEndpoints(self.emgapi_resource)]
except (KeyError, ValueError) as exc:
logger.warning(
f"Resource {self.emgapi_resource} does not have a defined access identifier key: {exc}"
)
return "id" # TODO
__all__ = ["DescribeEmgapiModule"]
