Add an optional arg deadline to dev.deprecated to raise warning after deadline

docs/monty.dev.md

@@ -18,27 +18,14 @@ with a possible replacement.
  * **replacement** (*callable*) – A replacement class or method.
  * **message** (*str*) – A warning message to be displayed.
  * **category** (*Warning*) – Choose the category of the warning to issue. Defaults
- to FutureWarning. Another choice can be DeprecationWarning. NOte that
+ to FutureWarning. Another choice can be DeprecationWarning. Note that
  FutureWarning is meant for end users and is always shown unless silenced.
  DeprecationWarning is meant for developers and is never shown unless
  python is run in developmental mode or the filter is changed. Make
  the choice accordingly.
 * **Returns**
  Original function, but with a warning to use the updated class.
 
-## monty.dev.get_ncpus()
-
-**NOTE**: If you are using Python >= 2.7, multiprocessing.cpu_count() already
-provides the number of CPUs. In fact, this is the first method tried.
-The purpose of this function is to cater to old Python versions that
-still exist on many Linux style clusters.
-
-Number of virtual or physical CPUs on this system, i.e.
-user/real as output by time(1) when called with an optimally scaling
-userspace-only program. Return -1 if ncpus cannot be detected. Taken from:
-[http://stackoverflow.com/questions/1006289/how-to-find-out-the-number-of](http://stackoverflow.com/questions/1006289/how-to-find-out-the-number-of)-
-cpus-in-python
-
 ## monty.dev.install_excepthook(hook_type=’color’, \*\*kwargs)
 
 This function replaces the original python traceback with an improved

docs/monty.md

@@ -44,7 +44,6 @@ useful design patterns such as singleton and cached_class, and many more.
  * `singleton()`
 * [monty.dev module](monty.dev.md)
  * `deprecated()`
- * `get_ncpus()`
  * `install_excepthook()`
  * `requires`
 * [monty.fnmatch module](monty.fnmatch.md)

monty/dev.py

@@ -5,21 +5,33 @@
 
 import functools
 import logging
+import os
 import sys
+import subprocess
 import warnings
+from datetime import datetime
+from typing import Callable, Optional, Type
 
 logger = logging.getLogger(__name__)
 
 
-def deprecated(replacement=None, message=None, category=FutureWarning):
+def deprecated(
+ replacement: Optional[Callable] = None,
+ message: str = "",
+ deadline: Optional[tuple[int, int, int]] = None,
+ category: Type[Warning] = FutureWarning,
+):
  """
  Decorator to mark classes or functions as deprecated, with a possible replacement.
 
  Args:
  replacement (callable): A replacement class or method.
  message (str): A warning message to be displayed.
+ deadline (Optional[tuple[int, int, int]]): Optional deadline for removal
+ of the old function/class, in format (yyyy, MM, dd). A CI warning would
+ be raised after this date if is running in code owner' repo.
  category (Warning): Choose the category of the warning to issue. Defaults
- to FutureWarning. Another choice can be DeprecationWarning. NOte that
+ to FutureWarning. Another choice can be DeprecationWarning. Note that
  FutureWarning is meant for end users and is always shown unless silenced.
  DeprecationWarning is meant for developers and is never shown unless
  python is run in developmental mode or the filter is changed. Make
@@ -29,8 +41,61 @@ def deprecated(replacement=None, message=None, category=FutureWarning):
  Original function, but with a warning to use the updated class.
  """
 
- def craft_message(old, replacement, message):
+ def _convert_date(date: tuple[int, int, int]) -> datetime:
+ """Convert date as int tuple (yyyy, MM, dd) to datetime type."""
+
+ return datetime(*date)
+
+ def raise_deadline_warning() -> None:
+ """Raise CI warning after removal deadline in code owner's repo."""
+
+ def _is_in_owner_repo() -> bool:
+ """Check if is running in code owner's repo.
+ Only generate reliable check when `git` is installed and remote name
+ is "origin".
+ """
+
+ try:
+ # Get current running repo
+ result = subprocess.run(
+ ["git", "config", "--get", "remote.origin.url"],
+ stdout=subprocess.PIPE,
+ )
+ owner_repo = (
+ result.stdout.decode("utf-8")
+ .strip()
+ .lstrip("https://github.com/") # https clone
+ .lstrip("[email protected]:") # ssh clone
+ .rstrip(".git") # ssh clone
+ )
+
+ return owner_repo == os.getenv("GITHUB_REPOSITORY")
+
+ except (subprocess.CalledProcessError, FileNotFoundError):
+ return False
+
+ # Only raise warning in code owner's repo CI
+ if (
+ _deadline is not None
+ and os.getenv("CI")
+ and datetime.now() > _deadline
+ and _is_in_owner_repo()
+ ):
+ raise DeprecationWarning(
+ "This function should have been removed on {deadline:%Y-%m-%d}."
+ )
+
+ def craft_message(
+ old: Callable,
+ replacement: Callable,
+ message: str,
+ deadline: datetime,
+ ):
  msg = f"{old.__name__} is deprecated"
+
+ if deadline is not None:
+ msg += f", and will be removed on {deadline:%Y-%m-%d}\n"
+
  if replacement is not None:
  if isinstance(replacement, property):
  r = replacement.fget
@@ -39,18 +104,25 @@ def craft_message(old, replacement, message):
  else:
  r = replacement
  msg += f"; use {r.__name__} in {r.__module__} instead."
- if message is not None:
+
+ if message:
  msg += "\n" + message
  return msg
 
- def deprecated_decorator(old):
+ def deprecated_decorator(old: Callable):
  def wrapped(*args, **kwargs):
- msg = craft_message(old, replacement, message)
+ msg = craft_message(old, replacement, message, _deadline)
  warnings.warn(msg, category=category, stacklevel=2)
  return old(*args, **kwargs)
 
  return wrapped
 
+ # Convert deadline to datetime
+ _deadline = _convert_date(deadline) if deadline is not None else None
+
+ # Raise a CI warning after removal deadline
+ raise_deadline_warning()
+
  return deprecated_decorator
 
 
@@ -101,7 +173,7 @@ def decorated(*args, **kwargs):
  return decorated
 
 
-def install_excepthook(hook_type="color", **kwargs):
+def install_excepthook(hook_type: str = "color", **kwargs):
  """
  This function replaces the original python traceback with an improved
  version from Ipython. Use `color` for colourful traceback formatting,

monty/json.py

@@ -328,7 +328,7 @@ def __get_pydantic_core_schema__(cls, source_type, handler):
  if core_schema is None:
  raise RuntimeError("Pydantic >= 2.0 is required for validation")
 
- s = core_schema.general_plain_validator_function(cls.validate_monty_v2)
+ s = core_schema.with_info_plain_validator_function(cls.validate_monty_v2)
 
  return core_schema.json_or_python_schema(json_schema=s, python_schema=s)
 

requirements-ci.txt

@@ -1,12 +1,9 @@
 pytest
 pytest-cov
 coverage
-coveralls
 pycodestyle
 mypy
-pydocstyle
 pydantic
-flake8
 black
 pylint
 torch

tests/test_dev.py

@@ -1,44 +1,33 @@
 import unittest
 import warnings
+import datetime
 
 import pytest
-
 from monty.dev import deprecated, install_excepthook, requires
 
-
-class A:
- @property
- def repl_prop(self):
- pass
-
- @deprecated(repl_prop) # type: ignore
- @property
- def prop(self):
- pass
+# Set all warnings to always be triggered.
+warnings.simplefilter("always")
 
 
 class TestDecorator:
  def test_deprecated(self):
- def func_a():
+ def func_replace():
  pass
 
- @deprecated(func_a, "hello")
- def func_b():
+ @deprecated(func_replace, "Use func_replace instead")
+ def func_old():
  pass
 
  with warnings.catch_warnings(record=True) as w:
- # Cause all warnings to always be triggered.
- warnings.simplefilter("always")
  # Trigger a warning.
- func_b()
- # Verify some things
+ func_old()
+ # Verify Warning and message
  assert issubclass(w[0].category, FutureWarning)
- assert "hello" in str(w[0].message)
+ assert "Use func_replace instead" in str(w[0].message)
 
  def test_deprecated_property(self):
- class a:
- def __init__(self):
- pass
+ class TestClass:
+ """A dummy class for tests."""
 
  @property
  def property_a(self):
@@ -54,58 +43,99 @@ def func_a(self):
  return "a"
 
  with warnings.catch_warnings(record=True) as w:
- # Cause all warnings to always be triggered.
- warnings.simplefilter("always")
  # Trigger a warning.
- assert a().property_b == "b"
- # Verify some things
+ assert TestClass().property_b == "b"
+ # Verify warning type
  assert issubclass(w[-1].category, FutureWarning)
 
  with warnings.catch_warnings(record=True) as w:
- # Cause all warnings to always be triggered.
- warnings.simplefilter("always")
  # Trigger a warning.
- assert a().func_a() == "a"
+ assert TestClass().func_a() == "a"
  # Verify some things
  assert issubclass(w[-1].category, FutureWarning)
 
  def test_deprecated_classmethod(self):
- class A:
- def __init__(self):
- pass
+ class TestClass:
+ """A dummy class for tests."""
 
  @classmethod
- def classmethod_a(self):
+ def classmethod_a(cls):
  pass
 
  @classmethod
  @deprecated(classmethod_a)
- def classmethod_b(self):
+ def classmethod_b(cls):
  return "b"
 
  with warnings.catch_warnings(record=True) as w:
- # Cause all warnings to always be triggered.
- warnings.simplefilter("always")
  # Trigger a warning.
- assert A().classmethod_b() == "b"
+ assert TestClass().classmethod_b() == "b"
  # Verify some things
  assert issubclass(w[-1].category, FutureWarning)
 
- class A:
- def __init__(self):
- pass
+ class TestClass_deprecationwarning:
+ """A dummy class for tests."""
 
  @classmethod
- def classmethod_a(self):
+ def classmethod_a(cls):
  pass
 
  @classmethod
  @deprecated(classmethod_a, category=DeprecationWarning)
- def classmethod_b(self):
+ def classmethod_b(cls):
  return "b"
 
  with pytest.warns(DeprecationWarning):
- assert A().classmethod_b() == "b"
+ assert TestClass_deprecationwarning().classmethod_b() == "b"
+
+ def test_deprecated_deadline(self):
+ @deprecated(deadline=(2000, 1, 1))
+ def func_old():
+ pass
+
+ with warnings.catch_warnings(record=True) as w:
+ # Trigger a warning.
+ func_old()
+ # Verify message
+ assert "will be removed on 2000-01-01" in str(w[0].message)
+
+ def test_deprecated_deadline_no_warn(self, monkeypatch):
+ # Test cases where no warning should be raised
+ @deprecated(deadline=(2000, 1, 1))
+ def func_old():
+ pass
+
+ # No warn case 1: date before deadline
+ with warnings.catch_warnings(record=True) as w:
+ monkeypatch.setattr(
+ datetime, "datetime", lambda: datetime.datetime(1999, 1, 1)
+ )
+ func_old()
+
+ for warning in w:
+ assert "This function should have been removed on" not in str(
+ warning.message
+ )
+
+ # No warn case 2: not in CI env
+ with warnings.catch_warnings(record=True) as w:
+ monkeypatch.delenv("CI", raising=False)
+ func_old()
+
+ for warning in w:
+ assert "This function should have been removed on" not in str(
+ warning.message
+ )
+
+ # No warn case 3: not in code owner repo
+ with warnings.catch_warnings(record=True) as w:
+ monkeypatch.setenv("GITHUB_REPOSITORY", "NONE/NONE")
+ func_old()
+
+ for warning in w:
+ assert "This function should have been removed on" not in str(
+ warning.message
+ )
 
  def test_requires(self):
  try: