🎨 New development decorators (#98)

Signed-off-by: Lukas Heumos <[email protected]>

Author: Zethson

.pre-commit-config.yaml

@@ -2,8 +2,8 @@ fail_fast: false
 default_language_version:
   python: python3
 default_stages:
-  - commit
-  - push
+  - pre-commit
+  - pre-push
 minimum_pre_commit_version: 2.16.0
 repos:
   - repo: https://github.com/pre-commit/mirrors-prettier
@@ -24,7 +24,7 @@ repos:
               docs/notes/
           )
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.5.5
+    rev: v0.9.1
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix, --unsafe-fixes]
@@ -44,7 +44,7 @@ repos:
       - id: trailing-whitespace
       - id: check-case-conflict
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.7.1
+    rev: v1.14.1
     hooks:
       - id: mypy
         args: [--no-strict-optional, --ignore-missing-imports]

lamin_utils/__init__.py

@@ -2,6 +2,9 @@
 
 __version__ = "0.13.10"
 
-from ._core import colors
+try:
+    from ._colors import colors
+except ImportError:  # Backward compatibility
+    from ._core import colors  # type: ignore
 from ._logger import logger
 from ._python_version import py_version_warning

lamin_utils/_core.py lamin_utils/_colors.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+# BSD 3-Clause License
+# Copyright (c) 2017-2018 P. Angerer, F. Alexander Wolf, Theis Lab
+# All rights reserved.
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+# * Redistributions of source code must retain the above copyright notice, this
+#   list of conditions and the following disclaimer.
+# * Redistributions in binary form must reproduce the above copyright notice,
+#   this list of conditions and the following disclaimer in the documentation
+#   and/or other materials provided with the distribution.
+# * Neither the name of the copyright holder nor the names of its
+#   contributors may be used to endorse or promote products derived from
+#   this software without specific prior written permission.
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+import warnings
+from functools import wraps
+
+
+def deprecated(new_name=None, remove_in_version=None):
+    """Decorator to mark functions as deprecated and hide them from docs.
+
+    This is a decorator which can be used to mark functions, methods and properties as deprecated.
+    It will result in a warning being emitted when the function is used.
+    It will also hide the function from the docs.
+
+    Args:
+        new_name: Name of the new function to use instead.
+            If `None`, omits the new name notice.
+        remove_in_version: Version when this will be removed.
+            If `None`, omits the remove in version X notice.
+
+    Example:
+        @property
+        @deprecated("n_files")
+        def n_objects(self) -> int:
+            return self.n_files
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            base_msg = f"{func.__name__} is deprecated"
+            version_msg = (
+                f" and will be removed in version {remove_in_version}"
+                if remove_in_version
+                else ""
+            )
+            migration_msg = f". Use {new_name} instead" if new_name else ""
+            msg = f"{base_msg}{version_msg}{migration_msg}."
+            warnings.warn(msg, category=DeprecationWarning, stacklevel=2)
+            return func(*args, **kwargs)
+
+        setattr(wrapper, "__deprecated", True)
+        return wrapper
+
+    return decorator
+
+
+def future_change(change_version=None, change_description=None):
+    """Warns about future behavior changes.
+
+    Args:
+        change_version: Version when behavior will change.
+        change_description: Description of the behavior change.
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            base_msg = f"{func.__name__} behavior will change"
+            version_msg = f" in version {change_version}" if change_version else ""
+            desc_msg = f". {change_description}" if change_description else ""
+            msg = f"{base_msg}{version_msg}{desc_msg}."
+            warnings.warn(msg, category=FutureWarning, stacklevel=2)
+            return func(*args, **kwargs)
+
+        setattr(wrapper, "__future_change", True)
+        return wrapper
+
+    return decorator
+
+
+def experimental(stable_version=None):
+    """Marks function as experimental/unstable API.
+
+    Args:
+        stable_version: Expected version for API stabilization.
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            base_msg = f"{func.__name__} is experimental"
+            version_msg = f" until version {stable_version}" if stable_version else ""
+            msg = f"{base_msg}{version_msg}. API may change without warning."
+            warnings.warn(msg, category=UserWarning, stacklevel=2)
+            return func(*args, **kwargs)
+
+        setattr(wrapper, "__experimental", True)
+        return wrapper
+
+    return decorator

lamin_utils/_compat.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Iterable
+from typing import TYPE_CHECKING
 
-from ._core import colors
+from ._colors import colors
 from ._logger import logger
 from ._map_synonyms import map_synonyms, to_str
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
+
     import numpy as np
     import pandas as pd
 
@@ -262,7 +264,7 @@ def _validate_logging(result: InspectResult, field: str | None = None) -> None:
         if len(result.non_validated) > 10:
             print_values += ", ..."
         warn_msg = (
-            f"{colors.yellow(f'{len(result.non_validated)} unique term{s}')} ({(100-result.frac_validated):.2f}%)"
+            f"{colors.yellow(f'{len(result.non_validated)} unique term{s}')} ({(100 - result.frac_validated):.2f}%)"
             f" {are} not validated{field_msg}: {colors.yellow(print_values)}"
         )
         if len(empty_warn_msg) > 0:

lamin_utils/_inspect.py

@@ -1,11 +1,16 @@
+from __future__ import annotations
+
 import re
 from collections import namedtuple
-from typing import Any, Dict, Iterable, List, Optional, Tuple
+from typing import TYPE_CHECKING, Any
 
 from ._logger import logger
 
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
 
-def _append_records_to_list(df_dict: Dict, value: str, record) -> None:
+def _append_records_to_list(df_dict: dict, value: str, record) -> None:
     """Append unique records to a list."""
     values_list = df_dict[value]
     if not isinstance(values_list, list):
@@ -20,19 +25,19 @@ def _append_records_to_list(df_dict: Dict, value: str, record) -> None:
 
 def _create_df_dict(
     df: Any = None,
-    field: Optional[str] = None,
-    records: Optional[List] = None,
-    values: Optional[List] = None,
-    tuple_name: Optional[str] = None,
-) -> Dict:
+    field: str | None = None,
+    records: list | None = None,
+    values: list | None = None,
+    tuple_name: str | None = None,
+) -> dict:
     """Create a dict with {lookup key: records in namedtuple}.
 
     Value is a list of namedtuples if multiple records match the same key.
     """
     if df is not None:
         records = df.itertuples(index=False, name=tuple_name)
         values = df[field]
-    df_dict: Dict = {}  # a dict of namedtuples as records and values as keys
+    df_dict: dict = {}  # a dict of namedtuples as records and values as keys
     for i, row in enumerate(records):  # type:ignore
         value = values[i]  # type:ignore
         if not isinstance(value, str):
@@ -52,12 +57,12 @@ class Lookup:
     # removed DataFrame type annotation to speed up import time
     def __init__(
         self,
-        field: Optional[str] = None,
+        field: str | None = None,
         tuple_name="MyTuple",
         prefix: str = "bt",
         df: Any = None,
-        values: Optional[Iterable] = None,
-        records: Optional[List] = None,
+        values: Iterable | None = None,
+        records: list | None = None,
     ) -> None:
         self._tuple_name = tuple_name
         if df is not None:
@@ -78,13 +83,13 @@ def __init__(
         self._lookup_dict = self._create_lookup_dict(lkeys=lkeys, df_dict=self._df_dict)
         self._prefix = prefix
 
-    def _to_lookup_keys(self, values: Iterable, prefix: str) -> Dict:
+    def _to_lookup_keys(self, values: Iterable, prefix: str) -> dict:
         """Convert a list of strings to tab-completion allowed formats.
 
         Returns:
             {lookup_key: value_or_values}
         """
-        lkeys: Dict = {}
+        lkeys: dict = {}
         for value in list(values):
             if not isinstance(value, str):
                 continue
@@ -103,8 +108,8 @@ def _to_lookup_keys(self, values: Iterable, prefix: str) -> Dict:
                 lkeys[lkey] = value
         return lkeys
 
-    def _create_lookup_dict(self, lkeys: Dict, df_dict: Dict) -> Dict:
-        lkey_dict: Dict = {}  # a dict of namedtuples as records and lookup keys as keys
+    def _create_lookup_dict(self, lkeys: dict, df_dict: dict) -> dict:
+        lkey_dict: dict = {}  # a dict of namedtuples as records and lookup keys as keys
         for lkey, values in lkeys.items():
             if isinstance(values, list):
                 combined_list = []
@@ -120,18 +125,18 @@ def _create_lookup_dict(self, lkeys: Dict, df_dict: Dict) -> Dict:
 
         return lkey_dict
 
-    def dict(self) -> Dict:
+    def dict(self) -> dict:
         """Dictionary of the lookup."""
         return self._df_dict
 
-    def lookup(self, return_field: Optional[str] = None) -> Tuple:
+    def lookup(self, return_field: str | None = None) -> tuple:
         """Lookup records with dot access."""
         # Names are invalid if they are conflict with Python keywords.
         if "class" in self._lookup_dict:
             self._lookup_dict[f"{self._prefix.lower()}_class"] = self._lookup_dict.pop(
                 "class"
             )
-        keys: List = list(self._lookup_dict.keys()) + ["dict"]
+        keys: list = list(self._lookup_dict.keys()) + ["dict"]
         MyTuple = namedtuple("Lookup", keys)  # type:ignore
         if return_field is not None:
             self._lookup_dict = {

lamin_utils/_lookup.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Iterable, Literal
+from typing import TYPE_CHECKING, Literal
 
 from ._logger import logger
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
+
     import pandas as pd
 
 

lamin_utils/_map_synonyms.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 from itertools import chain
-from typing import Any, Dict, Iterable, List, Literal, Union
+from typing import TYPE_CHECKING, Any, Literal
 
 from ._logger import logger
 from ._map_synonyms import map_synonyms
 
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
 
 def standardize(
     df: Any,