poc dataclass + custom methods

Author: orkes-harshil

src/conductor/client/http/models/workflow_run.py

@@ -1,7 +1,10 @@
 import pprint
 import re  # noqa: F401
+from dataclasses import dataclass, field, InitVar
+from typing import Dict, List, Optional, Any, ClassVar
 
 import six
+from deprecated import deprecated
 
 from conductor.client.http.models import Task
 
@@ -10,19 +13,14 @@
 running_status = ('RUNNING', 'PAUSED')
 
 
-class WorkflowRun(object):
+@dataclass
+class WorkflowRun:
     """NOTE: This class is auto generated by the swagger code generator program.
 
     Do not edit the class manually.
     """
-    """
-    Attributes:
-      swagger_types (dict): The key is attribute name
-                            and the value is attribute type.
-      attribute_map (dict): The key is attribute name
-                            and the value is json key in definition.
-    """
-    swagger_types = {
+    # Class attributes for swagger
+    swagger_types: ClassVar[Dict[str, str]] = {
         'correlation_id': 'str',
         'create_time': 'int',
         'created_by': 'str',
@@ -37,7 +35,7 @@ class WorkflowRun(object):
         'workflow_id': 'str'
     }
 
-    attribute_map = {
+    attribute_map: ClassVar[Dict[str, str]] = {
         'correlation_id': 'correlationId',
         'create_time': 'createTime',
         'created_by': 'createdBy',
@@ -52,6 +50,38 @@ class WorkflowRun(object):
         'workflow_id': 'workflowId'
     }
 
+    # Fields for the dataclass
+    correlation_id: Optional[str] = field(default=None)
+    create_time: Optional[int] = field(default=None)
+    created_by: Optional[str] = field(default=None)
+    input: Optional[Dict[str, Any]] = field(default=None)
+    output: Optional[Dict[str, Any]] = field(default=None)
+    priority: Optional[int] = field(default=None)
+    request_id: Optional[str] = field(default=None)
+    status: Optional[str] = field(default=None)
+    tasks: Optional[List[Task]] = field(default=None)
+    update_time: Optional[int] = field(default=None)
+    variables: Optional[Dict[str, Any]] = field(default=None)
+    workflow_id: Optional[str] = field(default=None)
+
+    # Additional init parameter (not a dataclass field)
+    reason_for_incompletion: InitVar[Optional[str]] = None
+
+    # Private backing fields
+    _correlation_id: Optional[str] = field(init=False, repr=False)
+    _create_time: Optional[int] = field(init=False, repr=False)
+    _created_by: Optional[str] = field(init=False, repr=False)
+    _input: Optional[Dict[str, Any]] = field(init=False, repr=False)
+    _output: Optional[Dict[str, Any]] = field(init=False, repr=False)
+    _priority: Optional[int] = field(init=False, repr=False)
+    _request_id: Optional[str] = field(init=False, repr=False)
+    _status: Optional[str] = field(init=False, repr=False)
+    _tasks: Optional[List[Task]] = field(init=False, repr=False)
+    _update_time: Optional[int] = field(init=False, repr=False)
+    _variables: Optional[Dict[str, Any]] = field(init=False, repr=False)
+    _workflow_id: Optional[str] = field(init=False, repr=False)
+    _reason_for_incompletion: Optional[str] = field(init=False, repr=False)
+
     def __init__(self, correlation_id=None, create_time=None, created_by=None, input=None, output=None, priority=None,
                  request_id=None, status=None, tasks=None, update_time=None, variables=None, workflow_id=None,
                  reason_for_incompletion: str = None):  # noqa: E501
@@ -95,6 +125,22 @@ def __init__(self, correlation_id=None, create_time=None, created_by=None, input
             self.workflow_id = workflow_id
         self._reason_for_incompletion = reason_for_incompletion
 
+    def __post_init__(self, reason_for_incompletion: Optional[str]):
+        """Initialize private fields after dataclass initialization"""
+        self._correlation_id = self.correlation_id
+        self._create_time = self.create_time
+        self._created_by = self.created_by
+        self._input = self.input
+        self._output = self.output
+        self._priority = self.priority
+        self._request_id = self.request_id
+        self._status = self.status
+        self._tasks = self.tasks
+        self._update_time = self.update_time
+        self._variables = self.variables
+        self._workflow_id = self.workflow_id
+        self._reason_for_incompletion = reason_for_incompletion
+
     @property
     def correlation_id(self):
         """Gets the correlation_id of this WorkflowRun.  # noqa: E501
@@ -442,4 +488,4 @@ def current_task(self) -> Task:
         for task in self.tasks:
             if task.status == 'SCHEDULED' or task.status == 'IN_PROGRESS':
                 current = task
-        return current
+        return current