mypy

Author: esoteric-ephemera

mp_api/client/contribs/_types.py

@@ -269,10 +269,13 @@ def from_file(cls, path: str | Path) -> Self:
             raise MPContribsClientError(f"use pathlib.Path or str (is: {typ}).")
 
         kind = guess(str(path))
-        supported = isinstance(kind, MPCC_SETTINGS.SUPPORTED_FILETYPES)
         content = path.read_bytes()
 
-        if not supported:  # try to gzip text file
+        if not (
+            supported := any(
+                isinstance(kind, ftyp) for ftyp in MPCC_SETTINGS.SUPPORTED_FILETYPES
+            )
+        ):  # try to gzip text file
             try:
                 content = gzip.compress(content)
             except Exception:

mp_api/client/contribs/client.py

@@ -54,7 +54,6 @@
     CONTRIBS_DOC_NAME,
     ContribData,
     ContribsProject,
-    ContribSubmission,
     QueryResult,
 )
 from mp_api.client.contribs.settings import MPCC_SETTINGS
@@ -115,7 +114,7 @@ def validate_email(email_string: str):
 
 
 def validate_url(
-    url_string: str, qualifying: Sequence[str, ...] = ("scheme", "netloc")
+    url_string: str, qualifying: Sequence[str] = ("scheme", "netloc")
 ) -> None:
     """Verify an endpoint URL.
 
@@ -722,7 +721,7 @@ def get_project(
         proj = self.projects.getProjectByName(pk=name, _fields=fields).result()
 
         return (
-            _convert_to_model(
+            _convert_to_model(  # type: ignore[return-value]
                 [proj],
                 ContribsProject,
                 model_name=CONTRIBS_DOC_NAME,
@@ -759,17 +758,7 @@ def query_projects(
         query = query or {}
 
         if self.project or "name" in query:
-            proj = self.get_project(name=query.get("name"), fields=fields)
-            return (
-                _convert_to_model(
-                    [proj],
-                    ContribsProject,
-                    model_name=CONTRIBS_DOC_NAME,
-                    requested_fields=fields,
-                )[0]
-                if self.use_document_model
-                else proj
-            )
+            return self.get_project(name=query.get("name"), fields=fields)
 
         if term:
 
@@ -798,7 +787,7 @@ def search_future(search_term):
 
         if total_pages < 2:
             return (
-                _convert_to_model(
+                _convert_to_model(  # type: ignore[return-value]
                     ret["data"],
                     ContribsProject,
                     model_name=CONTRIBS_DOC_NAME,
@@ -830,7 +819,7 @@ def search_future(search_term):
         ret["data"].extend([resp["result"]["data"] for resp in responses.values()])
 
         return (
-            _convert_to_model(
+            _convert_to_model(  # type: ignore[return-value]
                 ret["data"],
                 ContribsProject,
                 model_name=CONTRIBS_DOC_NAME,
@@ -858,7 +847,7 @@ def create_project(
                 raise MPContribsClientError(f"Project with {query} already exists!")
 
         project = ContribsProject(
-            **{
+            **{  # type: ignore[arg-type]
                 "name": name,
                 "title": title,
                 "authors": authors,
@@ -961,7 +950,7 @@ def delete_project(self, name: str | None = None) -> None:
 
     def get_contribution(
         self, cid: str, fields: list | None = None
-    ) -> MPCDict | ContribData:
+    ) -> ContribData | dict[str, Any]:  # type: ignore[return-value]
         """Retrieve a contribution.
 
         Args:
@@ -979,7 +968,7 @@ def get_contribution(
             pk=cid, _fields=fields
         ).result()
         return (
-            _convert_to_model(
+            _convert_to_model(  # type: ignore[return-value]
                 [contrib],
                 ContribData,
                 model_name=CONTRIBS_DOC_NAME,
@@ -1718,7 +1707,9 @@ def query_contributions(
             ret["data"] = [ContribData(**doc) for doc in ret["data"]]
 
         return (
-            _convert_to_model([ret], QueryResult, model_name=CONTRIBS_DOC_NAME)[0]
+            _convert_to_model(  # type: ignore[return-value]
+                [ret], QueryResult, model_name=CONTRIBS_DOC_NAME
+            )[0]
             if self.use_document_model
             else ret
         )
@@ -1884,7 +1875,7 @@ def _set_is_public(
 
     def submit_contributions(
         self,
-        contributions: list[dict | ContribSubmission],
+        contributions: list[dict],
         ignore_dupes: bool = False,
         timeout: int = -1,
         skip_dupe_check: bool = False,

mp_api/client/contribs/logger.py

@@ -72,12 +72,12 @@ def get_logger(name: str = "mp_api.client.contribs") -> CustomLoggerAdapter:
 
 
 class TqdmToLogger(StringIO):
-    logger: logging.Logger = MPCC_LOGGER
-    level: int = MPCC_SETTINGS.CLIENT_LOG_LEVEL
+    logger: logging.LoggerAdapter = MPCC_LOGGER
+    level: int | str = MPCC_SETTINGS.CLIENT_LOG_LEVEL
     buf: str = ""
 
     def __init__(
-        self, logger: logging.Logger = MPCC_LOGGER, level: int | None = None
+        self, logger: logging.LoggerAdapter = MPCC_LOGGER, level: int | None = None
     ) -> None:
         """Start an instance of a TQDM logger.
 

mp_api/client/contribs/schemas.py

@@ -40,6 +40,19 @@ def _cast_pandas_dtype(dtype: type, assume_nullable: bool = True) -> type:
 
 
 def _get_unit(v: str) -> tuple[str, str | None]:
+    """Parse a physical variable name and its optional unit from a string value.
+
+    Ex:
+        _get_unit("Temperature [K]") or _get_unit("Temperature (K)")
+        will both return "Temperature", "K"
+
+    Args:
+        v (str) : input column name to parse
+
+    Returns:
+        str : the base name of the physical quantity
+        str or None: its optional unit
+    """
     if (matched := re.search(r"\(([^)]+)\)|\[([^\]]+)\]", v)) is not None:
         try:
             unit_group = next(
@@ -54,13 +67,31 @@ def _get_unit(v: str) -> tuple[str, str | None]:
 
 
 def _to_camel_case(v: str) -> str:
+    """Convert a generic string to CamelCase.
+
+    Args:
+        v (str) : input string
+
+    Returns:
+        str : CamelCase string
+    """
     split_strs = [y for x in v.lower().split() for y in x.split("_")]
     return "".join(s[0].upper() + s[1:] for s in split_strs)
 
 
 def _get_pydantic_from_dataframe(
     df: pd.DataFrame,
 ) -> tuple[type[BaseModel], dict[str, str]]:
+    """Dynamically create a pydantic BaseModel from a pandas DataFrame.
+
+    Args:
+        df : pandas DataFrame to parse
+
+    Returns:
+        BaseModel: the inferred schema of the pandas DataFrame
+        dict of str to str: the remapped column names in an
+            MP Contribs compatible format.
+    """
     columns_renamed = {}
     columns_to_unit = {}
     for col in df.columns:
@@ -85,28 +116,28 @@ def _get_pydantic_from_dataframe(
         if not all(pd.isna(df[col_name]))
     }
 
-    return (
-        create_model(
-            "InferredModel",
-            **model_fields,
-        ),
-        columns_renamed,
-    )
+    return create_model("InferredModel", **model_fields), columns_renamed
 
 
 class Reference(_DictLikeAccess):
+    """Define schema of URL reference."""
+
     label: str
     url: str
 
 
 class Column(_DictLikeAccess):
+    """Define schema of MP Contribs column statistics."""
+
     path: str
     min: float | None = float("nan")
     max: float | None = float("nan")
     unit: str = "NaN"
 
 
 class Stats(_DictLikeAccess):
+    """Define aggregated project statistics schema."""
+
     columns: int = 0
     contributions: int = 0
     tables: int = 0
@@ -116,6 +147,8 @@ class Stats(_DictLikeAccess):
 
 
 class ContribsProject(_DictLikeAccess):
+    """Define schema for MP Contribs Project."""
+
     name: str | None = None
     title: str | None = None
     authors: str | None = None
@@ -134,12 +167,14 @@ class ContribsProject(_DictLikeAccess):
 
     @field_validator("other", mode="before")
     def flatten_other(cls, d: dict) -> dict[str, str | None]:
+        """Flatten column metadata."""
         if all(isinstance(v, str) for v in d.values()):
             return d
         return flatten_dict(d)
 
     @field_serializer("other", mode="plain")
     def unflatten_other(self, v: dict[str, str]) -> dict[str, Any]:
+        """Unflatten column metadata."""
         return unflatten_dict(v)
 
 

mp_api/client/core/schemas.py

@@ -37,11 +37,12 @@ def __str__(self) -> str:
         populated_fields = sorted(
             k for k in self.__class__.model_fields if getattr(self, k, None) is not None
         )
-        annos = {k: self.__class__.model_fields[k].annotation for k in populated_fields}
-        for k, v in annos.items():
-            if hasattr(v, "__name__"):
-                annos[k] = v.__name__
-        # return f"{self.__class__.__name__}({len(populated_fields)} populated fields)"
+        _annos: dict[str, Any] = {
+            k: self.__class__.model_fields[k].annotation for k in populated_fields
+        }
+        annos: dict[str, str] = {
+            k: getattr(v, "__name__", str(v)) for k, v in _annos.items()
+        }
         return (
             f"{self.__class__.__name__}(\n"
             + "\n".join(

mp_api/client/mprester.py

@@ -139,7 +139,7 @@ def __init__(
             local_dataset_cache: Target directory for downloading full datasets. Defaults
                 to "mp_datasets" in the user's home directory
             force_renew: Option to overwrite existing local dataset
-            **kwargs: access to legacy kwargs that may be in the process of being deprecated
+            **kwargs: access to ContribsClient kwargs or (possibly-deprecated) legacy kwargs
         """
         self.api_key = get_user_api_key(api_key=api_key)
 
@@ -159,6 +159,14 @@ def __init__(
         self.local_dataset_cache = local_dataset_cache
         self.force_renew = force_renew
         self._contribs = None
+        self._contribs_kwargs = {
+            k: kwargs[k]
+            for k in (
+                "host",
+                "project",
+            )
+            if k in kwargs
+        }
 
         self._deprecated_attributes = [
             "eos",
@@ -239,7 +247,16 @@ def __init__(
                 )
 
     @property
-    def contribs(self):
+    def contribs(self) -> None:
+        """Create an instance of the MP ContribsClient.
+
+        NB: The following args are taken from `MPRester`:
+            - api_key
+            - headers
+            - session
+            - use_document_model
+
+        """
         if self._contribs is None:
             try:
                 from mp_api.client.contribs.client import ContribsClient
@@ -249,6 +266,7 @@ def contribs(self):
                     headers=self.headers,
                     session=self.session,
                     use_document_model=self.use_document_model,
+                    **self._contribs_kwargs,
                 )
 
             except ImportError:
@@ -265,7 +283,6 @@ def contribs(self):
                     category=MPRestWarning,
                     stacklevel=2,
                 )
-
         return self._contribs
 
     def __enter__(self):

pyproject.toml

@@ -131,6 +131,7 @@ isort.required-imports = ["from __future__ import annotations"]
 [tool.mypy]
 namespace_packages = true
 ignore_missing_imports = true
+follow_untyped_imports = true
 
 [tool.coverage.report]
 exclude_also = [