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

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:
@@ -930,6 +934,34 @@ def get_type_name(tp: Type[CfgVal]) -> str:
         return str(tp)
 
 
+U = TypeVar("U", bound="StructuredRunOpt")
+
+
+class StructuredRunOpt(abc.ABC):
+
+    @abstractmethod
+    def template(self) -> str:
+        """
+        Returns the template string for the StructuredRunOpt
+        """
+        pass
+
+    def __repr__(self) -> str:
+        return self.template().format(**asdict(self))
+
+    def __eq__(self, other: Type[T]) -> 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)
+
+
 @dataclass
 class runopt:
     """

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,40 @@ 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)
+
+
 class RunConfigTest(unittest.TestCase):
     def get_cfg(self) -> Mapping[str, CfgVal]:
         return {