Refactoring HistoryEvent

Get rid of all of those hardcoded property value references

azure/durable_functions/interfaces/__init__.py

@@ -1,8 +0,0 @@
-"""Interfaces for durable functions."""
-from .IAction import IAction
-from .IFunctionContext import IFunctionContext
-
-__all__ = [
-    'IAction',
-    'IFunctionContext'
-]

azure/durable_functions/models/DurableOrchestrationContext.py

@@ -1,10 +1,7 @@
 import json
-import logging
 import datetime
 from typing import List, Any, Dict
 
-from dateutil.parser import parse as dt_parse
-
 from . import (RetryOptions)
 from .history import HistoryEvent, HistoryEventType
 from ..interfaces import IAction
@@ -22,7 +19,7 @@ class DurableOrchestrationContext:
     def __init__(self,
                  context_string: str):
         context: Dict[str, Any] = json.loads(context_string)
-        self._histories: List[HistoryEvent] = context.get("history")
+        self._histories: List[HistoryEvent] = [HistoryEvent(**he) for he in context.get("history")]
         self._instance_id = context.get("instanceId")
         self._is_replaying = context.get("isReplaying")
         self._parent_instance_id = context.get("parentInstanceId")
@@ -38,10 +35,10 @@ def __init__(self,
                 input_=i)
         self.task_all = lambda t: task_all(tasks=t)
         self.decision_started_event: HistoryEvent = list(filter(
-            lambda e_: e_["EventType"] == HistoryEventType.ORCHESTRATOR_STARTED,
+            lambda e_: e_.event_type == HistoryEventType.ORCHESTRATOR_STARTED,
             self.histories))[0]
         self._current_utc_datetime = \
-            dt_parse(self.decision_started_event["Timestamp"])
+            self.decision_started_event.timestamp
         self.new_guid_counter = 0
         self.actions: List[List[IAction]] = []
 

azure/durable_functions/models/OrchestratorState.py

@@ -2,6 +2,7 @@
 from typing import List, Any, Dict
 
 from .utils.json_utils import add_attrib
+from ..interfaces.IAction import IAction
 
 
 class OrchestratorState:
@@ -13,7 +14,7 @@ class OrchestratorState:
 
     def __init__(self,
                  is_done: bool,
-                 actions: List[List[Any]],
+                 actions: List[List[IAction]],
                  output: Any,
                  error: str = None,
                  custom_status: Any = None):
@@ -24,7 +25,7 @@ def __init__(self,
         self._custom_status: Any = custom_status
 
     @property
-    def actions(self) -> List[List[Any]]:
+    def actions(self) -> List[List[IAction]]:
         """Get the ordered list of async actions the orchestrator function should perform.
 
         This list is append-only; it must contain all scheduled async actions up to the latest

azure/durable_functions/models/Task.py

@@ -1,4 +1,6 @@
 from datetime import datetime
+
+from tests.test_utils.constants import DATETIME_STRING_FORMAT
 from ..interfaces import IAction
 
 
@@ -56,6 +58,9 @@ def timestamp(self) -> datetime:
         """Get the timestamp of the task."""
         return self._timestamp
 
+    def timestamp_as_string(self):
+        return self._timestamp.strftime(DATETIME_STRING_FORMAT)
+
     @property
     def id(self):
         """Get the ID number of the task.

azure/durable_functions/models/TaskSet.py

@@ -1,5 +1,5 @@
 from typing import List
-from ..interfaces import IAction
+from ..interfaces.IAction import IAction
 
 
 class TaskSet:
@@ -36,7 +36,7 @@ def is_faulted(self) -> bool:
         return self._is_faulted
 
     @property
-    def actions(self) -> IAction:
+    def actions(self) -> List[IAction]:
         """Get the scheduled action represented by the task.
 
         _Internal use only._

azure/durable_functions/models/history/HistoryEvent.py

@@ -1,12 +1,49 @@
+import datetime
+from dateutil.parser import parse as dt_parse
 from .HistoryEventType import HistoryEventType
 
 
 class HistoryEvent:
     """Used to communicate state relevant information from the durable extension to the client."""
 
-    def __init__(self):
-        self.EventType: HistoryEventType
-        self.EventId: int
-        self.IsPlayed: bool
-        self.Timestamp: str
-        self.IsProcessed: bool = False
+    # noinspection PyPep8Naming
+    def __init__(self, EventType: HistoryEventType, EventId: int, IsPlayed: bool, Timestamp: str,
+                 **kwargs):
+        self._event_type: HistoryEventType = EventType
+        self._event_id: int = EventId
+        self._is_played: bool = IsPlayed
+        self._timestamp: datetime = dt_parse(Timestamp)
+        self._is_processed: bool = False
+        if kwargs is not None:
+            [self.__setattr__(key, value) for key, value in kwargs.items()]
+
+    @property
+    def event_type(self) -> HistoryEventType:
+        """Get the history event type
+
+        Returns
+        ----------
+        HistoryEventType: The type of history event
+
+        """
+        return self._event_type
+
+    @property
+    def event_id(self) -> int:
+        return self._event_id
+
+    @property
+    def is_played(self) -> bool:
+        return self._is_played
+
+    @property
+    def is_processed(self) -> bool:
+        return self._is_processed
+
+    @is_processed.setter
+    def is_processed(self, value: bool):
+        self._is_processed = value
+
+    @property
+    def timestamp(self) -> datetime:
+        return self._timestamp

azure/durable_functions/models/utils/json_utils.py

@@ -1,32 +1,54 @@
 from typing import Dict, Any
 
+from tests.test_utils.constants import DATETIME_STRING_FORMAT
+
 
 def add_attrib(json_dict: Dict[str, Any], object_,
                attribute_name: str, alt_name: str = None):
     """Add the value of the attribute from the object to the dictionary.
 
     Used to dynamically add the value of the attribute if the value is present.
 
-    :param json_dict: The dictionary to add the attribute to
-    :param object_: The object to look for the attribute on
-    :param attribute_name: The name of the attribute to look for
-    :param alt_name: An alternate name to provide to the attribute in the in the dictionary
+    Parameters
+    ----------
+    json_dict: The dictionary to add the attribute to
+    object_: The object to look for the attribute on
+    attribute_name: The name of the attribute to look for
+    alt_name: An alternate name to provide to the attribute in the in the dictionary
     """
     if hasattr(object_, attribute_name):
         json_dict[alt_name or attribute_name] = \
             getattr(object_, attribute_name)
 
 
+def add_datetime_attrib(json_dict: Dict[str, Any], object_,
+                        attribute_name: str, alt_name: str = None):
+    """Add the value of the attribute from the object to the dictionary converted into a string.
+
+    Parameters
+    ----------
+    json_dict: The dictionary to add the attribute to
+    object_: The object to look for the attribute on
+    attribute_name: The name of the attribute to look for
+    alt_name: An alternate name to provide to the attribute in the in the dictionary
+    """
+    if hasattr(object_, attribute_name):
+        json_dict[alt_name or attribute_name] = \
+            getattr(object_, attribute_name).strftime(DATETIME_STRING_FORMAT)
+
+
 def add_json_attrib(json_dict: Dict[str, Any], object_,
                     attribute_name: str, alt_name: str = None):
     """Add the results of the to_json() function call of the attribute from the object to the dict.
 
     Used to dynamically add the JSON converted value of the attribute if the value is present.
 
-    :param json_dict: The dictionary to add the attribute to
-    :param object_: The object to look for the attribute on
-    :param attribute_name: The name of the attribute to look for
-    :param alt_name: An alternate name to provide to the attribute in the in the dictionary
+    Parameters
+    ----------
+    json_dict: The dictionary to add the attribute to
+    object_: The object to look for the attribute on
+    attribute_name: The name of the attribute to look for
+    alt_name: An alternate name to provide to the attribute in the in the dictionary
     """
     if hasattr(object_, attribute_name):
         json_dict[alt_name or attribute_name] = getattr(object_, attribute_name).to_json()

azure/durable_functions/orchestrator.py

@@ -3,13 +3,9 @@
 Responsible for orchestrating the execution of the user defined generator
 function.
 """
-import logging
-import traceback
 from typing import Callable, Iterator, Any
 
-from dateutil.parser import parse as dt_parse
-
-from .interfaces import IFunctionContext
+from .interfaces.IFunctionContext import IFunctionContext
 from .models import (
     DurableOrchestrationContext,
     Task,
@@ -37,7 +33,7 @@ def __init__(self,
         self.fn: Callable[[IFunctionContext], Iterator[Any]] = activity_func
         self.customStatus: Any = None
 
-    # noinspection PyAttributeOutsideInit
+    # noinspection PyAttributeOutsideInit,PyUnboundLocalVariable
     def handle(self, context_string: str):
         """Handle the orchestration of the user defined generator function.
 
@@ -86,8 +82,6 @@ def handle(self, context_string: str):
                 actions=self.durable_context.actions,
                 custom_status=self.customStatus)
         except Exception as e:
-            e_string = traceback.format_exc()
-            logging.warning(f"!!!Generator Termination Exception {e_string}")
             orchestration_state = OrchestratorState(
                 is_done=False,
                 output=None,  # Should have no output, after generation range
@@ -113,20 +107,19 @@ def _add_to_actions(self, generation_state):
             self.durable_context.actions.append(generation_state.actions)
 
     def _reset_timestamp(self):
-        last_timestamp = dt_parse(
-            self.durable_context.decision_started_event['Timestamp'])
+        last_timestamp = self.durable_context.decision_started_event.timestamp
         decision_started_events = list(
             filter(lambda e_: (
-                e_["EventType"] == HistoryEventType.ORCHESTRATOR_STARTED
-                and dt_parse(e_["Timestamp"]) > last_timestamp),
+                e_.event_type == HistoryEventType.ORCHESTRATOR_STARTED
+                and e_.timestamp > last_timestamp),
                 self.durable_context.histories))
         if len(decision_started_events) == 0:
             self.durable_context.current_utc_datetime = None
         else:
             self.durable_context.decision_started_event = \
                 decision_started_events[0]
-            self.durable_context.current_utc_datetime = dt_parse(
-                self.durable_context.decision_started_event['Timestamp'])
+            self.durable_context.current_utc_datetime = \
+                self.durable_context.decision_started_event.timestamp
 
     @classmethod
     def create(cls, fn):

azure/durable_functions/tasks/call_activity.py

@@ -35,19 +35,19 @@ def call_activity_task(
             is_faulted=False,
             action=new_action,
             result=parse_history_event(task_completed),
-            timestamp=task_completed["Timestamp"],
-            id_=task_completed["TaskScheduledId"])
+            timestamp=task_completed.timestamp,
+            id_=task_completed.TaskScheduledId)
 
     if task_failed is not None:
         return Task(
             is_completed=True,
             is_faulted=True,
             action=new_action,
-            result=task_failed["Reason"],
-            timestamp=task_failed["Timestamp"],
-            id_=task_failed["TaskScheduledId"],
+            result=task_failed.Reason,
+            timestamp=task_failed.timestamp,
+            id_=task_failed.TaskScheduledId,
             exc=Exception(
-                f"{task_failed['Reason']} \n {task_failed['Details']}")
+                f"{task_failed.Reason} \n {task_failed.Details}")
         )
 
     return Task(is_completed=False, is_faulted=False, action=new_action)

azure/durable_functions/tasks/call_activity_with_retry.py

@@ -48,20 +48,19 @@ def call_activity_with_retry_task(
                 is_faulted=False,
                 action=new_action,
                 result=parse_history_event(task_completed),
-                timestamp=task_completed["Timestamp"],
-                id_=task_completed["TaskScheduledId"])
+                timestamp=task_completed.timestamp,
+                id_=task_completed.TaskScheduledId)
 
         if task_failed and task_retry_timer and attempt + 1 >= \
                 retry_options.max_number_of_attempts:
             return Task(
                 is_completed=True,
                 is_faulted=True,
                 action=new_action,
-                result=task_failed["Reason"],
-                timestamp=task_failed["Timestamp"],
-                id_=task_failed["TaskScheduledId"],
+                timestamp=task_failed.timestamp,
+                id_=task_failed.TaskScheduledId,
                 exc=Exception(
-                    f"{task_failed['Reason']} \n {task_failed['Details']}")
+                    f"{task_failed.Reason} \n {task_failed.Details}")
             )
 
     return Task(is_completed=False, is_faulted=False, action=new_action)