Promote to master

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 All notable changes to this project will be documented in this file.
 
+## 1.0.0b6
+
+- [Create timer](https://github.com/Azure/azure-functions-durable-python/issues/35) functionality available
+
+## 1.0.0b5
+
+- [Object serialization](https://github.com/Azure/azure-functions-durable-python/issues/90) made available
+- [Can set custom status](https://github.com/Azure/azure-functions-durable-python/issues/117) of orchestration
+
 ## 1.0.0b3-b4
 - Release to test CD pipeline with push to PyPI
 

azure/durable_functions/models/DurableOrchestrationContext.py

@@ -9,7 +9,7 @@
 from ..models.Task import Task
 from ..models.TokenSource import TokenSource
 from ..tasks import call_activity_task, task_all, task_any, call_activity_with_retry_task, \
-    wait_for_external_event_task, continue_as_new, new_uuid, call_http
+    wait_for_external_event_task, continue_as_new, new_uuid, call_http, create_timer_task
 from azure.functions._durable_functions import _deserialize_custom_object
 
 
@@ -51,6 +51,7 @@ def __init__(self,
         self.continue_as_new = lambda i: continue_as_new(input_=i)
         self.task_any = lambda t: task_any(tasks=t)
         self.task_all = lambda t: task_all(tasks=t)
+        self.create_timer = lambda d: create_timer_task(state=self.histories, fire_at=d)
         self.decision_started_event: HistoryEvent = \
             [e_ for e_ in self.histories
              if e_.event_type == HistoryEventType.ORCHESTRATOR_STARTED][0]

azure/durable_functions/models/actions/CreateTimerAction.py

@@ -0,0 +1,42 @@
+from typing import Any, Dict
+
+from .ActionType import ActionType
+from ..utils.json_utils import add_attrib, add_datetime_attrib
+import datetime
+
+
+class CreateTimerAction:
+    """Defines the structure of the Create Timer object.
+
+    Returns
+    -------
+        Information needed by durable extension to schedule the activity
+
+    Raises
+    ------
+    ValueError
+        if the event fired is not of valid datetime object
+    """
+
+    def __init__(self, fire_at: datetime, is_cancelled: bool = False):
+        self.action_type: ActionType = ActionType.CREATE_TIMER
+        self.fire_at: datetime = fire_at
+        self.is_cancelled: bool = is_cancelled
+
+        if not isinstance(self.fire_at, datetime.date):
+            raise ValueError("fireAt: Expected valid datetime object but got ", self.fire_at)
+
+    def to_json(self) -> Dict[str, Any]:
+        """
+        Convert object into a json dictionary.
+
+        Returns
+        -------
+        Dict[str, Any]
+            The instance of the class converted into a json dictionary
+        """
+        json_dict = {}
+        add_attrib(json_dict, self, 'action_type', 'actionType')
+        add_datetime_attrib(json_dict, self, 'fire_at', 'fireAt')
+        add_attrib(json_dict, self, 'is_cancelled', 'isCanceled')
+        return json_dict

azure/durable_functions/models/actions/__init__.py

@@ -5,12 +5,14 @@
 from .CallActivityWithRetryAction import CallActivityWithRetryAction
 from .WaitForExternalEventAction import WaitForExternalEventAction
 from .CallHttpAction import CallHttpAction
+from .CreateTimerAction import CreateTimerAction
 
 __all__ = [
     'Action',
     'ActionType',
     'CallActivityAction',
     'CallActivityWithRetryAction',
     'CallHttpAction',
-    'WaitForExternalEventAction'
+    'WaitForExternalEventAction',
+    'CreateTimerAction'
 ]

azure/durable_functions/tasks/__init__.py

@@ -8,6 +8,7 @@
 from .continue_as_new import continue_as_new
 from .new_uuid import new_uuid
 from .call_http import call_http
+from .create_timer import create_timer_task
 
 __all__ = [
     'call_activity_task',
@@ -18,5 +19,6 @@
     'task_all',
     'task_any',
     'should_suspend',
-    'wait_for_external_event_task'
+    'wait_for_external_event_task',
+    'create_timer_task'
 ]

azure/durable_functions/tasks/create_timer.py

@@ -0,0 +1,41 @@
+from typing import List
+from ..models.actions.CreateTimerAction import CreateTimerAction
+from ..models.history import HistoryEvent
+from .task_utilities import find_task_timer_created, find_task_retry_timer_fired, set_processed
+import datetime
+from .timer_task import TimerTask
+
+
+def create_timer_task(state: List[HistoryEvent],
+                      fire_at: datetime) -> TimerTask:
+    """Durable Timers are used in orchestrator function to implement delays.
+
+    Parameters
+    ----------
+    state : List[HistoryEvent]
+        The list of history events to search to determine the current state of the activity
+    fire_at : datetime
+        The time interval to fire the timer trigger
+
+    Returns
+    -------
+    TimerTask
+        A Durable Timer Task that schedules the timer to wake up the activity
+    """
+    new_action = CreateTimerAction(fire_at)
+
+    timer_created = find_task_timer_created(state, fire_at)
+    timer_fired = find_task_retry_timer_fired(state, timer_created)
+
+    set_processed([timer_created, timer_fired])
+
+    if timer_fired:
+        return TimerTask(
+            is_completed=True, action=new_action,
+            timestamp=timer_fired.timestamp,
+            id_=timer_fired.event_id)
+    else:
+        return TimerTask(
+            is_completed=False, action=new_action,
+            timestamp=None,
+            id_=None)

azure/durable_functions/tasks/task_utilities.py

@@ -1,5 +1,6 @@
 import json
 from ..models.history import HistoryEventType
+from ..constants import DATETIME_STRING_FORMAT
 from azure.functions._durable_functions import _deserialize_custom_object
 
 
@@ -118,6 +119,29 @@ def find_task_failed(state, scheduled_task):
     return tasks[0]
 
 
+def find_task_timer_created(state, fire_at):
+    """Locate the Timer Created Task.
+
+    Within the state passed, search for an event that has hasn't been processed,
+    is a timer created task type,
+    and has the an event id that is one higher then Scheduled Id of the provided
+    failed task provided.
+    """
+    if fire_at is None:
+        return None
+
+    tasks = []
+    for e in state:
+        if e.event_type == HistoryEventType.TIMER_CREATED and hasattr(e, "FireAt"):
+            if e.FireAt == fire_at.strftime(DATETIME_STRING_FORMAT):
+                tasks.append(e)
+
+    if len(tasks) == 0:
+        return None
+
+    return tasks[0]
+
+
 def find_task_retry_timer_created(state, failed_task):
     """Locate the Timer Created Task.
 

azure/durable_functions/tasks/timer_task.py

@@ -0,0 +1,47 @@
+from ..models.Task import Task
+
+
+class TimerTask(Task):
+    """Represents a pending timer.
+
+    All pending timers must be completed or canceled for an orchestration to complete.
+
+    Example: Cancel a timer
+    ```
+    timeout_task = context.df.create_timer(expiration_date)
+    if not timeout_task.is_completed():
+        timeout_task.cancel()
+    ```
+    """
+
+    def __init__(self, action, is_completed, timestamp, id_):
+        self._action = action
+        self._is_completed = is_completed
+        self._timestamp = timestamp
+        self._id = id_
+
+        super().__init__(self._is_completed, False,
+                         self._action, None, self._timestamp, self._id, None)
+
+    def is_cancelled(self) -> bool:
+        """Check of a timer is cancelled.
+
+        Returns
+        -------
+        bool
+            Returns whether a timer has been cancelled or not
+        """
+        return self._action.is_cancelled
+
+    def cancel(self):
+        """Cancel a timer.
+
+        Raises
+        ------
+        ValueError
+            Raises an error if the task is already completed and an attempt is made to cancel it
+        """
+        if not self._is_completed:
+            self._action.is_cancelled = True
+        else:
+            raise ValueError("Cannot cancel a completed task.")

samples/aml_monitoring/.funcignore

@@ -0,0 +1,5 @@
+.git*
+.vscode
+local.settings.json
+test
+py36

samples/aml_monitoring/.gitignore

@@ -0,0 +1,44 @@
+bin
+obj
+csx
+.vs
+edge
+Publish
+
+*.user
+*.suo
+*.cscfg
+*.Cache
+project.lock.json
+
+/packages
+/TestResults
+
+/tools/NuGet.exe
+/App_Data
+/secrets
+/data
+.secrets
+appsettings.json
+local.settings.json
+
+node_modules
+dist
+
+# Local python packages
+.python_packages/
+
+# Python Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+py36

samples/aml_monitoring/.vscode/extensions.json

@@ -0,0 +1,5 @@
+{
+    "recommendations": [
+        "ms-azuretools.vscode-azurefunctions"
+    ]
+}

samples/aml_monitoring/aml_durable_orchestrator/__init__.py

@@ -0,0 +1,45 @@
+import logging,json
+import azure.durable_functions as df
+from datetime import datetime,timedelta
+
+def orchestrator_fn(context: df.DurableOrchestrationContext):
+    pipeline_endpoint = ""
+    experiment_name = ""
+
+    # Step 1:  Kickoff the AML pipeline
+    input_args= {}
+    input_args["pipeline_endpoint"] = pipeline_endpoint
+    input_args["experiment_name"] = experiment_name
+    input_args["params"] = None
+    run_id = yield context.call_activity("aml_pipeline",input_args)
+    polling_interval = 60
+    expiry_time = context.current_utc_datetime + timedelta(minutes=30)
+
+    # Consider continueAsNew - use this in the samples
+    # while loop explodes the history table on high scale
+    while context.current_utc_datetime < expiry_time:
+
+    # Step 2: Poll the status of the pipeline
+        poll_args = {}
+        poll_args["run_id"] = run_id
+        poll_args["experiment_name"] = experiment_name
+        job_status = yield context.call_activity("aml_poll_status",poll_args)
+
+        # Use native Dictionary fix the generic binding conversion in worker. Can it return a Dict?
+        activity_status = json.loads(job_status)
+        if activity_status["status_code"] == 202:
+            next_check = context.current_utc_datetime + timedelta(minutes=1)
+
+            # Set intermediate status for anyone who wants to poll this durable function
+            context.set_custom_status(activity_status)
+
+            yield context.create_timer(next_check)
+
+        elif activity_status["status_code"] == 500:
+            job_completed = True
+            raise Exception("AML Job Failed/Cancelled...")
+        else:
+            job_completed = True
+    return activity_status
+
+main = df.Orchestrator.create(orchestrator_fn)

samples/aml_monitoring/aml_durable_orchestrator/function.json

@@ -0,0 +1,11 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "name": "context",
+      "type": "orchestrationTrigger",
+      "direction": "in"
+    }
+  ],
+  "disabled": false
+}

samples/aml_monitoring/aml_pipeline/__init__.py

@@ -0,0 +1,33 @@
+import logging,json
+import os
+import time
+from typing import Dict
+import requests
+import azure.functions as func
+from azureml.core.authentication import ServicePrincipalAuthentication
+
+from ..shared.auth_helper import get_access_token
+
+
+def trigger_aml_endpoint(pipeline_endpoint, experiment_name, parameter_body, retries=3):
+    aad_token = get_access_token()
+    response = requests.post(
+        pipeline_endpoint,
+        headers=aad_token,
+        json={"ExperimentName": experiment_name,
+            "ParameterAssignments": parameter_body})
+
+    if response.status_code == 200:
+        success = True
+
+    return json.loads(response.content)
+
+# explicitly typing input_args causes exception
+def main(name):
+    input_args = json.loads(name)
+    try:
+        response = trigger_aml_endpoint(input_args["pipeline_endpoint"], input_args["experiment_name"], input_args["params"])
+    except Exception as exception:
+        logging.error("Got exception: ", exc_info=True)
+        return exception
+    return response["Id"]