Introduce StructuredRunOpt for more complex runopts (#1154)

Summary:

StructuredRunOpt can be used to create new complex runopts

This can be used as for example:

```
    dataclass
    class UlimitTest(StructuredRunOpt):
        name: str
        hard: int
        soft: int

        def template(self) -> str:
            return "{name},{soft:d},{hard:d}"
```
This comes with 
1. template() that helps the from_repr() use that template to map to the fields with types
2. __eq__ to check equality between two

Differential Revision: D85159071

Author: AbishekS

docs/source/specs.rst

@@ -52,6 +52,9 @@ Run Configs
 .. autoclass:: runopts
    :members:
 
+.. autoclass:: StructuredRunOpt
+   :members:
+
 Run Status
 --------------
 .. autoclass:: AppStatus

requirements.txt

@@ -4,3 +4,4 @@ docker
 filelock
 fsspec>=2023.10.0
 tabulate
+parse

torchx/specs/api.py

@@ -6,6 +6,7 @@
 
 # pyre-strict
 
+import abc
 import asyncio
 import copy
 import inspect
@@ -17,6 +18,7 @@
 import shutil
 import typing
 import warnings
+from abc import abstractmethod
 from dataclasses import asdict, dataclass, field
 from datetime import datetime
 from enum import Enum, IntEnum
@@ -39,6 +41,8 @@
     Union,
 )
 
+import parse
+
 from torchx.util.types import to_dict
 
 _APP_STATUS_FORMAT_TEMPLATE = """AppStatus:
@@ -877,11 +881,69 @@ def __init__(self, status: AppStatus, *args: object) -> None:
         self.status = status
 
 
+U = TypeVar("U", bound="StructuredRunOpt")
+
+
+class StructuredRunOpt(abc.ABC):
+
+    @abstractmethod
+    def template(self) -> str:
+        """
+        Returns the template string for the StructuredRunOpt.
+        The template string should contain the field names of the StructuredRunOpt object.
+
+        Usage
+
+        .. code-block:: python
+
+            @dataclass
+            class Ulimit(StructuredRunOpt):
+                name: str
+                hard: int
+                soft: int
+
+                def template(self) -> str:
+                    return "{name},{soft:d},{hard:d}"
+
+            opts = runopts()
+            opts.add("ulimit", type_=self.Ulimit, help="ulimits for the container")
+            cfg = opts.resolve(
+                {
+                    "ulimit": self.Ulimit.from_repr(
+                        "test,50,100",
+                    )
+                }
+            )
+
+
+        The template string for the Ulimit class is "{name},{soft:d},{hard:d}".
+        These are mapped to the field names of the Ulimit object.
+        Template strings also may need types as in the above example :d for int.
+        As seen in the usage, .from_repr() is used to create a Ulimit object from a string representation.
+        """
+        pass
+
+    def __repr__(self) -> str:
+        return self.template().format(**asdict(self))
+
+    def __eq__(self, other: Any) -> bool:
+        return isinstance(other, type(self)) and asdict(self) == asdict(other)
+
+    @classmethod
+    def from_repr(cls: Type[U], repr: str) -> U:
+        """
+        Parses the repr string and returns a StructuredRunOpt object
+        """
+        tmpl = cls.__new__(cls).template()
+        result = parse.parse(tmpl, repr)
+        return cls(**result.named)
+
+
 # valid run cfg values; only support primitives (str, int, float, bool, List[str], Dict[str, str])
 # TODO(wilsonhong): python 3.9+ supports list[T] in typing, which can be used directly
 # in isinstance(). Should replace with that.
 # see: https://docs.python.org/3/library/stdtypes.html#generic-alias-type
-CfgVal = Union[str, int, float, bool, List[str], Dict[str, str], None]
+CfgVal = Union[str, int, float, bool, List[str], Dict[str, str], StructuredRunOpt, None]
 
 
 T = TypeVar("T")

torchx/specs/test/api_test.py

@@ -14,7 +14,7 @@
 import time
 import unittest
 import warnings
-from dataclasses import asdict
+from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import Dict, List, Mapping, Tuple, Union
 from unittest import mock
@@ -43,6 +43,7 @@
     RoleStatus,
     runopt,
     runopts,
+    StructuredRunOpt,
     TORCHX_HOME,
     Workspace,
 )
@@ -550,6 +551,56 @@ def test_getset_metadata(self) -> None:
         self.assertEqual(None, app.metadata.get("non_existent"))
 
 
+class StructuredRunOptTest(unittest.TestCase):
+
+    @dataclass
+    class UlimitTest(StructuredRunOpt):
+        name: str
+        hard: int
+        soft: int
+
+        def template(self) -> str:
+            return "{name},{soft:d},{hard:d}"
+
+    def test_structured_runopt(self) -> None:
+        opt = self.UlimitTest(name="test", hard=100, soft=50)
+
+        # Test class from_repr
+        self.assertEqual(
+            opt,
+            self.UlimitTest.from_repr(
+                "test,50,100",
+            ),
+        )
+
+        # Test repr
+        self.assertEqual(
+            "StructuredRunOptTest.UlimitTest(name='test', hard=100, soft=50)", repr(opt)
+        )
+
+        # Test equality
+        opt_other = self.UlimitTest(name="test", hard=100, soft=50)
+        self.assertEqual(opt, opt_other)
+        opt_other = self.UlimitTest(name="test", hard=100, soft=70)
+        self.assertNotEqual(opt, opt_other)
+
+        # Test with runopts
+
+        opts = runopts()
+        opts.add("ulimit", type_=self.UlimitTest, help="test ulimit")
+        cfg = opts.resolve(
+            {
+                "ulimit": self.UlimitTest.from_repr(
+                    "test,50,100",
+                )
+            }
+        )
+        self.assertEqual(
+            cfg.get("ulimit"),
+            self.UlimitTest(name="test", hard=100, soft=50),
+        )
+
+
 class RunConfigTest(unittest.TestCase):
     def get_cfg(self) -> Mapping[str, CfgVal]:
         return {