Fix pydantic issues with hera.workflows.Env (#505)

Signed-off-by: GitHub <[email protected]>
Signed-off-by: Sambhav Kothari <[email protected]>

docs/examples/workflows/upstream/colored_logs.md

@@ -0,0 +1,68 @@
+# Colored Logs
+
+> Note: This example is a replication of an Argo Workflow example in Hera. The upstream example can be [found here](https://github.com/argoproj/argo-workflows/blob/master/examples/colored-logs.yaml).
+
+
+
+## Hera
+
+```python
+from hera.workflows import Env, Script, Workflow
+
+
+def colored_logs():
+    import time  # noqa: I001
+    import random
+
+    messages = [
+        "No Color",
+        "\x1b[30m%s\x1b[0m" % "FG Black",
+        "\x1b[32m%s\x1b[0m" % "FG Green",
+        "\x1b[34m%s\x1b[0m" % "FG Blue",
+        "\x1b[36m%s\x1b[0m" % "FG Cyan",
+        "\x1b[41m%s\x1b[0m" % "BG Red",
+        "\x1b[43m%s\x1b[0m" % "BG Yellow",
+        "\x1b[45m%s\x1b[0m" % "BG Magenta",
+    ]
+    for i in range(1, 100):
+        print(random.choice(messages))
+        time.sleep(1)
+
+
+with Workflow(generate_name="colored-logs-") as w:
+    s = Script(
+        name="whalesay",
+        source=colored_logs,
+        image="python:3.7",
+        add_cwd_to_sys_path=False,
+        env=[Env(name="PYTHONUNBUFFERED", value="1")],
+    )
+    w.entrypoint = s.name
+```
+
+## YAML
+
+```yaml
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: colored-logs-
+spec:
+  entrypoint: whalesay
+  templates:
+  - name: whalesay
+    script:
+      command:
+      - python
+      env:
+      - name: PYTHONUNBUFFERED
+        value: '1'
+      image: python:3.7
+      source: "import time  # noqa: I001\nimport random\n\nmessages = [\n    \"No\
+        \ Color\",\n    \"\\x1b[30m%s\\x1b[0m\" % \"FG Black\",\n    \"\\x1b[32m%s\\\
+        x1b[0m\" % \"FG Green\",\n    \"\\x1b[34m%s\\x1b[0m\" % \"FG Blue\",\n   \
+        \ \"\\x1b[36m%s\\x1b[0m\" % \"FG Cyan\",\n    \"\\x1b[41m%s\\x1b[0m\" % \"\
+        BG Red\",\n    \"\\x1b[43m%s\\x1b[0m\" % \"BG Yellow\",\n    \"\\x1b[45m%s\\\
+        x1b[0m\" % \"BG Magenta\",\n]\nfor i in range(1, 100):\n    print(random.choice(messages))\n\
+        \    time.sleep(1)\n"
+```

examples/workflows/upstream/colored-logs.upstream.yaml

@@ -0,0 +1,32 @@
+# This example demonstrates colored logs of the logs viewer on Argo UI.
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: colored-logs-
+spec:
+  entrypoint: whalesay
+  templates:
+  - name: whalesay
+    script:
+      image: python:3.7
+      command: ["python"]
+      source: |
+        import time
+        import random
+
+        messages = [
+            'No Color',
+            '\x1b[30m%s\x1b[0m' % 'FG Black',
+            '\x1b[32m%s\x1b[0m' % 'FG Green',
+            '\x1b[34m%s\x1b[0m' % 'FG Blue',
+            '\x1b[36m%s\x1b[0m' % 'FG Cyan',
+            '\x1b[41m%s\x1b[0m' % 'BG Red',
+            '\x1b[43m%s\x1b[0m' % 'BG Yellow',
+            '\x1b[45m%s\x1b[0m' % 'BG Magenta',
+        ]
+        for i in range(1, 100):
+          print(random.choice(messages))
+          time.sleep(1)
+      env:
+      - name: PYTHONUNBUFFERED
+        value: '1'

examples/workflows/upstream/colored-logs.yaml

@@ -0,0 +1,22 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: colored-logs-
+spec:
+  entrypoint: whalesay
+  templates:
+  - name: whalesay
+    script:
+      command:
+      - python
+      env:
+      - name: PYTHONUNBUFFERED
+        value: '1'
+      image: python:3.7
+      source: "import time  # noqa: I001\nimport random\n\nmessages = [\n    \"No\
+        \ Color\",\n    \"\\x1b[30m%s\\x1b[0m\" % \"FG Black\",\n    \"\\x1b[32m%s\\\
+        x1b[0m\" % \"FG Green\",\n    \"\\x1b[34m%s\\x1b[0m\" % \"FG Blue\",\n   \
+        \ \"\\x1b[36m%s\\x1b[0m\" % \"FG Cyan\",\n    \"\\x1b[41m%s\\x1b[0m\" % \"\
+        BG Red\",\n    \"\\x1b[43m%s\\x1b[0m\" % \"BG Yellow\",\n    \"\\x1b[45m%s\\\
+        x1b[0m\" % \"BG Magenta\",\n]\nfor i in range(1, 100):\n    print(random.choice(messages))\n\
+        \    time.sleep(1)\n"

examples/workflows/upstream/colored_logs.py

@@ -0,0 +1,31 @@
+from hera.workflows import Env, Script, Workflow
+
+
+def colored_logs():
+    import time  # noqa: I001
+    import random
+
+    messages = [
+        "No Color",
+        "\x1b[30m%s\x1b[0m" % "FG Black",
+        "\x1b[32m%s\x1b[0m" % "FG Green",
+        "\x1b[34m%s\x1b[0m" % "FG Blue",
+        "\x1b[36m%s\x1b[0m" % "FG Cyan",
+        "\x1b[41m%s\x1b[0m" % "BG Red",
+        "\x1b[43m%s\x1b[0m" % "BG Yellow",
+        "\x1b[45m%s\x1b[0m" % "BG Magenta",
+    ]
+    for i in range(1, 100):
+        print(random.choice(messages))
+        time.sleep(1)
+
+
+with Workflow(generate_name="colored-logs-") as w:
+    s = Script(
+        name="whalesay",
+        source=colored_logs,
+        image="python:3.7",
+        add_cwd_to_sys_path=False,
+        env=[Env(name="PYTHONUNBUFFERED", value="1")],
+    )
+    w.entrypoint = s.name

src/hera/workflows/env.py

@@ -23,10 +23,6 @@
 class _BaseEnv(_BaseModel):
     name: str
 
-    def __init__(self, name: str):
-        self.name = name
-        super().__init__()
-
     def build(self) -> _ModelEnvVar:
         raise NotImplementedError
 
@@ -35,13 +31,6 @@ class Env(_BaseEnv):
     value: Optional[Any] = None
     value_from_input: Optional[Union[str, Parameter]] = None
 
-    def __init__(
-        self, name: str, value: Optional[Any] = None, value_from_input: Optional[Union[str, Parameter]] = None
-    ):
-        self.value = value
-        self.value_from_input = value_from_input
-        super().__init__(name)
-
     @staticmethod
     def _sanitise_param_for_argo(v: str) -> str:
         """Argo has some strict parameter validation. To satisfy, we replace all ._ with a dash,
@@ -54,6 +43,7 @@ def _sanitise_param_for_argo(v: str) -> str:
         return f"{legit_prefix}-{hash_suffix}"
 
     @root_validator(pre=True)
+    @classmethod
     def _check_values(cls, values):
         if values.get("value") is not None and values.get("value_from_input") is not None:
             raise ValueError("cannot specify both `value` and `value_from_input`")
@@ -84,12 +74,6 @@ class SecretEnv(_BaseEnv):
     secret_name: Optional[str] = None
     optional: Optional[bool] = None
 
-    def __init__(self, name: str, secret_key: str, secret_name: Optional[str] = None, optional: Optional[bool] = None):
-        self.secret_key = secret_key
-        self.secret_name = secret_name
-        self.optional = optional
-        super().__init__(name)
-
     def build(self) -> _ModelEnvVar:
         """Constructs and returns the Argo environment specification"""
         return _ModelEnvVar(
@@ -107,14 +91,6 @@ class ConfigMapEnv(_BaseEnv):
     config_map_name: Optional[str]
     optional: Optional[bool] = None
 
-    def __init__(
-        self, name: str, config_map_key: str, config_map_name: Optional[str] = None, optional: Optional[bool] = None
-    ):
-        self.config_map_key = config_map_key
-        self.config_map_name = config_map_name
-        self.optional = optional
-        super().__init__(name)
-
     def build(self) -> _ModelEnvVar:
         """Constructs and returns the Argo environment specification"""
         return _ModelEnvVar(
@@ -131,12 +107,8 @@ class FieldEnv(_BaseEnv):
     field_path: str
     api_version: Optional[str] = None
 
-    def __init__(self, name: str, field_path: str, api_version: Optional[str] = None):
-        self.field_path = field_path
-        self.api_version = api_version
-        super().__init__(name)
-
     @validator("api_version")
+    @classmethod
     def _check_api_version(cls, v):
         if v is None:
             return global_config.api_version
@@ -160,14 +132,6 @@ class ResourceEnv(_BaseEnv):
     container_name: Optional[str] = None
     divisor: Optional[Quantity] = None
 
-    def __init__(
-        self, name: str, resource: str, container_name: Optional[str] = None, divisor: Optional[Quantity] = None
-    ):
-        self.resource = resource
-        self.container_name = container_name
-        self.divisor = divisor
-        super().__init__(name)
-
     def build(self) -> _ModelEnvVar:
         return _ModelEnvVar(
             name=self.name,