Add deprecated_aliases to runopt and add warning (#1142)

Summary:


Similar to `RunOptAlias` lets introduce and use `RunOptDeprecatedAlias`. This will warn the user with a`UserWarning` when the user uses that specific name and suggests the primary one instead.

Differential Revision: D84180061

Author: AbishekS

torchx/specs/api.py

@@ -15,6 +15,7 @@
 import pathlib
 import re
 import typing
+import warnings
 from dataclasses import asdict, dataclass, field
 from datetime import datetime
 from enum import Enum
@@ -890,6 +891,10 @@ class RunOptAlias(str):
     pass
 
 
+class RunOptDeprecatedAlias(str):
+    pass
+
+
 @dataclass
 class runopt:
     """
@@ -901,6 +906,7 @@ class runopt:
     is_required: bool
     help: str
     aliases: list[RunOptAlias] | None = None
+    deprecated_aliases: list[RunOptDeprecatedAlias] | None = None
 
     @property
     def is_type_list_of_str(self) -> bool:
@@ -992,7 +998,7 @@ class runopts:
 
     def __init__(self) -> None:
         self._opts: Dict[str, runopt] = {}
-        self._alias_to_key: Dict[RunOptAlias, str] = {}
+        self._alias_to_key: Dict[RunOptAlias | RunOptDeprecatedAlias, str] = {}
 
     def __iter__(self) -> Iterator[Tuple[str, runopt]]:
         return self._opts.items().__iter__()
@@ -1043,12 +1049,24 @@ def resolve(self, cfg: Mapping[str, CfgVal]) -> Dict[str, CfgVal]:
             val = resolved_cfg.get(cfg_key)
             resolved_name = None
             aliases = runopt.aliases or []
+            deprecated_aliases = runopt.deprecated_aliases or []
             if val is None:
                 for alias in aliases:
                     val = resolved_cfg.get(alias)
                     if val is not None:
                         resolved_name = alias
                         break
+                for alias in deprecated_aliases:
+                    val = resolved_cfg.get(alias)
+                    if val is not None:
+                        resolved_name = alias
+                        use_instead = self._alias_to_key.get(alias)
+                        warnings.warn(
+                            f"Run option: {alias}, is deprecated. Please use {use_instead} instead",
+                            UserWarning,
+                            stacklevel=2,
+                        )
+                        break
             else:
                 resolved_name = cfg_key
                 for alias in aliases:
@@ -1174,20 +1192,23 @@ def cfg_from_json_repr(self, json_repr: str) -> Dict[str, CfgVal]:
     def _get_primary_key_and_aliases(
         self,
         cfg_key: list[str] | str,
-    ) -> tuple[str, list[RunOptAlias]]:
+    ) -> tuple[str, list[RunOptAlias], list[RunOptDeprecatedAlias]]:
         """
         Returns the primary key and aliases for the given cfg_key.
         """
         if isinstance(cfg_key, str):
-            return cfg_key, []
+            return cfg_key, [], []
 
         if len(cfg_key) == 0:
             raise ValueError("cfg_key must be a non-empty list")
         primary_key = None
         aliases = list[RunOptAlias]()
+        deprecated_aliases = list[RunOptDeprecatedAlias]()
         for name in cfg_key:
             if isinstance(name, RunOptAlias):
                 aliases.append(name)
+            elif isinstance(name, RunOptDeprecatedAlias):
+                deprecated_aliases.append(name)
             else:
                 if primary_key is not None:
                     raise ValueError(
@@ -1198,7 +1219,7 @@ def _get_primary_key_and_aliases(
             raise ValueError(
                 "Missing cfg_key. Please provide one other than the aliases."
             )
-        return primary_key, aliases
+        return primary_key, aliases, deprecated_aliases
 
     def add(
         self,
@@ -1213,7 +1234,9 @@ def add(
         value (if any). If the ``default`` is not specified then this option
         is a required option.
         """
-        primary_key, aliases = self._get_primary_key_and_aliases(cfg_key)
+        primary_key, aliases, deprecated_aliases = self._get_primary_key_and_aliases(
+            cfg_key
+        )
         if required and default is not None:
             raise ValueError(
                 f"Required option: {cfg_key} must not specify default value. Given: {default}"
@@ -1224,9 +1247,11 @@ def add(
                     f"Option: {cfg_key}, must be of type: {type_}."
                     f" Given: {default} ({type(default).__name__})"
                 )
-        opt = runopt(default, type_, required, help, aliases)
+        opt = runopt(default, type_, required, help, aliases, deprecated_aliases)
         for alias in aliases:
             self._alias_to_key[alias] = primary_key
+        for deprecated_alias in deprecated_aliases:
+            self._alias_to_key[deprecated_alias] = primary_key
         self._opts[primary_key] = opt
 
     def update(self, other: "runopts") -> None:

torchx/specs/test/api_test.py

@@ -13,6 +13,7 @@
 import tempfile
 import time
 import unittest
+import warnings
 from dataclasses import asdict
 from pathlib import Path
 from typing import Dict, List, Mapping, Tuple, Union
@@ -42,6 +43,7 @@
     RoleStatus,
     runopt,
     RunOptAlias,
+    RunOptDeprecatedAlias,
     runopts,
     TORCHX_HOME,
     Workspace,
@@ -602,6 +604,24 @@ def test_runopts_resolve_with_aliases(self) -> None:
         with self.assertRaises(InvalidRunConfigException):
             opts.resolve({"job_priority": "high", "jobPriority": "low"})
 
+    def test_runopts_add_with_deprecated_aliases(self) -> None:
+        opts = runopts()
+        opts.add(
+            [RunOptDeprecatedAlias("jobPriority"), "job_priority"],
+            type_=str,
+            help="run as user",
+        )
+        opts.resolve({"job_priority": "high"})
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            opts.resolve({"jobPriority": "high"})
+            self.assertEqual(len(w), 1)
+            self.assertEqual(w[0].category, UserWarning)
+            self.assertEqual(
+                str(w[0].message),
+                "Run option: jobPriority, is deprecated. Please use job_priority instead",
+            )
+
     def get_runopts(self) -> runopts:
         opts = runopts()
         opts.add("run_as", type_=str, help="run as user", required=True)