add .flake file and fix flake8 errors

.flake8

@@ -0,0 +1,8 @@
+[flake8]
+# delete D100~D107 for docstring checks
+# W503 contradicts with pep8 and will soon be fixed by flake8
+ignore = W503, D100, D101, D102, D103, D104, D107
+max-line-length = 99
+exclude =
+    __pycache__,
+    azure/durable_functions/grpc/protobuf/

azure-pipelines.yml

@@ -32,7 +32,8 @@ steps:
   displayName: 'Autogenerate gRPC Python files'
 
 - script: |
-    flake8 . --count --show-source --statistics --exit-zero 
+    cd azure
+    flake8 . --count --show-source --statistics
   displayName: 'Run lint test with flake8'
 
 - script: |

azure/durable_functions/__init__.py

@@ -4,4 +4,4 @@
 __all__ = [
     'Orchestrator',
     'DurableOrchestrationClient'
-]
+]

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -25,7 +25,8 @@ def __init__(self, context: str):
         self._showHistoryQueryKey: str = "showHistory"
         self._showHistoryOutputQueryKey: str = "showHistoryOutput"
         self._showInputQueryKey: str = "showInput"
-        self._orchestrationBindings: DurableOrchestrationBindings = DurableOrchestrationBindings(context)
+        self._orchestrationBindings: DurableOrchestrationBindings = \
+            DurableOrchestrationBindings(context)
 
     def start_new(self,
                   orchestration_function_name: str,
@@ -42,7 +43,8 @@ def get_json_input(client_input):
 
     def get_start_new_url(self, instance_id, orchestration_function_name):
         request_url = self._orchestrationBindings.creation_urls['createNewInstancePostUri']
-        request_url = request_url.replace(self._functionNamePlaceholder, orchestration_function_name)
+        request_url = request_url.replace(self._functionNamePlaceholder,
+                                          orchestration_function_name)
         request_url = request_url.replace(self._instanceIdPlaceholder,
                                           f'/{instance_id}' if instance_id is not None else '')
         return request_url

azure/durable_functions/models/OrchestratorState.py

@@ -1,6 +1,7 @@
 import json
 from typing import List, Any, Dict
-from .utils.json_utils import add_attrib, add_json_attrib
+from .utils.json_utils import add_attrib
+
 
 class OrchestratorState:
     def __init__(self,

azure/durable_functions/models/RetryOptions.py

@@ -15,6 +15,7 @@ def __init__(self, first_retry_interval_in_milliseconds: int, max_number_of_atte
     def to_json(self) -> Dict[str, Any]:
         json_dict = {}
 
-        add_attrib(json_dict, self, 'first_retry_interval_in_milliseconds', 'firstRetryIntervalInMilliseconds')
+        add_attrib(json_dict, self, 'first_retry_interval_in_milliseconds',
+                   'firstRetryIntervalInMilliseconds')
         add_attrib(json_dict, self, 'max_number_of_attempts', 'maxNumberOfAttempts')
         return json_dict

azure/durable_functions/models/actions/CallActivityAction.py

@@ -1,7 +1,7 @@
 from typing import Any, Dict
 
 from .ActionType import ActionType
-from ..utils.json_utils import add_attrib, add_json_attrib
+from ..utils.json_utils import add_attrib
 
 
 class CallActivityAction:

azure/durable_functions/models/history/HistoryEvent.py

@@ -1,4 +1,3 @@
-from datetime import datetime
 from .HistoryEventType import HistoryEventType
 
 

azure/durable_functions/models/history/__init__.py

@@ -4,4 +4,4 @@
 __all__ = [
     'HistoryEvent',
     'HistoryEventType'
-]
+]

azure/durable_functions/models/utils/json_utils.py

@@ -8,4 +8,4 @@ def add_attrib(json_dict: Dict[str, Any], object_, attribute_name: str, alt_name
 
 def add_json_attrib(json_dict: Dict[str, Any], object_, attribute_name: str, alt_name: str = None):
     if hasattr(object_, attribute_name):
-        json_dict[alt_name or attribute_name] = getattr(object_, attribute_name).to_json()
+        json_dict[alt_name or attribute_name] = getattr(object_, attribute_name).to_json()

azure/durable_functions/orchestrator.py

@@ -88,14 +88,15 @@ def _reset_timestamp(self):
         last_timestamp = dt_parse(self.durable_context.decision_started_event['Timestamp'])
         decision_started_events = list(
             filter(lambda e_: (
-                    e_["EventType"] == HistoryEventType.OrchestratorStarted
-                    and dt_parse(e_["Timestamp"]) > last_timestamp),
-                   self.durable_context.histories))
+                e_["EventType"] == HistoryEventType.OrchestratorStarted
+                and dt_parse(e_["Timestamp"]) > last_timestamp),
+                self.durable_context.histories))
         if len(decision_started_events) == 0:
             self.durable_context.currentUtcDateTime = None
         else:
             self.durable_context.decision_started_event = decision_started_events[0]
-            self.durable_context.currentUtcDateTime = dt_parse(self.durable_context.decision_started_event['Timestamp'])
+            self.durable_context.currentUtcDateTime = \
+                dt_parse(self.durable_context.decision_started_event['Timestamp'])
 
     @classmethod
     def create(cls, fn):

azure/durable_functions/tasks/__init__.py

@@ -8,4 +8,4 @@
     'call_activity_with_retry_task',
     'task_all',
     'should_suspend'
-]
+]

azure/durable_functions/tasks/call_activity.py

@@ -5,8 +5,8 @@
     Task)
 from ..models.actions.CallActivityAction import CallActivityAction
 from ..models.history import HistoryEvent
-from .task_utilities import find_task_completed, find_task_failed, find_task_scheduled, set_processed, \
-    parse_history_event
+from .task_utilities import find_task_completed, find_task_failed, find_task_scheduled, \
+    set_processed, parse_history_event
 
 
 def call_activity_task(

azure/durable_functions/tasks/call_activity_with_retry.py

@@ -1,26 +1,31 @@
+import logging
 from typing import List, Any
 
 from ..models.Task import (
     Task)
 from ..models.actions.CallActivityWithRetryAction import CallActivityWithRetryAction
 from ..models.history import HistoryEvent
 from ..models.RetryOptions import RetryOptions
-from .task_utilities import *
+from .task_utilities import (find_task_scheduled, find_task_completed, find_task_failed,
+                             find_task_retry_timer_created, find_task_retry_timer_fired,
+                             set_processed, parse_history_event)
 
 
 def call_activity_with_retry_task(
         state: List[HistoryEvent],
         retry_options: RetryOptions,
         name: str,
         input_: Any = None) -> Task:
-    new_action = CallActivityWithRetryAction(function_name=name, retry_options=retry_options, input_=input_)
+    new_action = CallActivityWithRetryAction(function_name=name,
+                                             retry_options=retry_options, input_=input_)
     for attempt in range(retry_options.max_number_of_attempts):
         task_scheduled = find_task_scheduled(state, name)
         task_completed = find_task_completed(state, task_scheduled)
         task_failed = find_task_failed(state, task_scheduled)
         task_retry_timer = find_task_retry_timer_created(state, task_failed)
         task_retry_timer_fired = find_task_retry_timer_fired(state, task_retry_timer)
-        set_processed([task_scheduled, task_completed, task_failed, task_retry_timer, task_retry_timer_fired])
+        set_processed([task_scheduled, task_completed, task_failed, task_retry_timer,
+                      task_retry_timer_fired])
 
         if not task_scheduled:
             break
@@ -35,7 +40,8 @@ def call_activity_with_retry_task(
                 timestamp=task_completed["Timestamp"],
                 id=task_completed["TaskScheduledId"])
 
-        if task_failed and task_retry_timer and attempt + 1 >= retry_options.max_number_of_attempts:
+        if task_failed and task_retry_timer \
+                and attempt + 1 >= retry_options.max_number_of_attempts:
             logging.warning("!!!Task Failed")
             return Task(
                 isCompleted=True,

azure/durable_functions/tasks/task_utilities.py

@@ -30,8 +30,9 @@ def find_task_scheduled(state, name):
 
     tasks = list(
         filter(lambda e: not (
-                (not (e["EventType"] == HistoryEventType.TaskScheduled) or not (e["Name"] == name)) or e.get(
-            "IsProcessed")), state))
+                (not (e["EventType"] == HistoryEventType.TaskScheduled)
+                    or not (e["Name"] == name))
+            or e.get("IsProcessed")), state))
 
     logging.warning(f"!!! findTaskScheduled {tasks}")
     if len(tasks) == 0:
@@ -46,8 +47,7 @@ def find_task_completed(state, scheduled_task):
 
     tasks = list(
         filter(lambda e: not (not (e["EventType"] == HistoryEventType.TaskCompleted) or not (
-                e.get("TaskScheduledId") == scheduled_task["EventId"])),
-               state))
+            e.get("TaskScheduledId") == scheduled_task["EventId"])), state))
 
     if len(tasks) == 0:
         return None
@@ -61,8 +61,7 @@ def find_task_failed(state, scheduled_task):
 
     tasks = list(
         filter(lambda e: not (not (e["EventType"] == HistoryEventType.TaskFailed) or not (
-                e.get("TaskScheduledId") == scheduled_task["EventId"])),
-               state))
+            e.get("TaskScheduledId") == scheduled_task["EventId"])), state))
 
     if len(tasks) == 0:
         return None
@@ -76,8 +75,7 @@ def find_task_retry_timer_created(state, failed_task):
 
     tasks = list(
         filter(lambda e: not (not (e["EventType"] == HistoryEventType.TimerCreated) or not (
-                e.get("EventId") == failed_task["TaskScheduledId"] + 1)),
-               state))
+            e.get("EventId") == failed_task["TaskScheduledId"] + 1)), state))
 
     if len(tasks) == 0:
         return None
@@ -91,8 +89,7 @@ def find_task_retry_timer_fired(state, retry_timer_created):
 
     tasks = list(
         filter(lambda e: not (not (e["EventType"] == HistoryEventType.TimerFired) or not (
-                e.get("TimerId") == retry_timer_created["EventId"])),
-               state))
+            e.get("TimerId") == retry_timer_created["EventId"])), state))
 
     if len(tasks) == 0:
         return None

samples/python_durable_bindings/DurableActivity/__init__.py

@@ -1,5 +1,6 @@
 import logging
 
+
 def main(name: str) -> str:
     logging.warning(f"Activity Triggered: {name}")
-    return f'Hello Activity: {name}!'
+    return f'Hello Activity: {name}!'

samples/python_durable_bindings/DurableFanoutOrchestrationTrigger/__init__.py

@@ -26,4 +26,8 @@ def main(context: str):
 
 
 if __name__ == "__main__":
-    main('{"history":[{"EventType":12,"EventId":-1,"IsPlayed":false,"Timestamp":"2019-12-08T23:18:41.3240927Z"},{"OrchestrationInstance":{"InstanceId":"48d0f95957504c2fa579e810a390b938","ExecutionId":"fd183ee02e4b4fd18c95b773cfb5452b"},"EventType":0,"ParentInstance":null,"Name":"DurableFunctionsOrchestratorJS","Version":"","Input":"null","Tags":null,"EventId":-1,"IsPlayed":false,"Timestamp":"2019-12-08T23:18:39.756132Z"}],"input":null,"instanceId":"48d0f95957504c2fa579e810a390b938","isReplaying":false,"parentInstanceId":null}')
+    main('{"history":[{"EventType":12,"EventId":-1,"IsPlayed":false,"Timestamp":"2019-12-08T23:18:41.3240927Z"},\
+        {"OrchestrationInstance":{"InstanceId":"48d0f95957504c2fa579e810a390b938","ExecutionId":"fd183ee02e4b4fd18c95b773cfb5452b"},\
+        "EventType":0,"ParentInstance":null,"Name":"DurableFunctionsOrchestratorJS","Version":"","Input":"null","Tags":null,"EventId":-1,\
+        "IsPlayed":false,"Timestamp":"2019-12-08T23:18:39.756132Z"}],"input":null,"instanceId":"48d0f95957504c2fa579e810a390b938",\
+        "isReplaying":false,"parentInstanceId":null}')

setup.py

@@ -5,7 +5,7 @@
 import sys
 import glob
 
-from setuptools import setup,find_packages
+from setuptools import setup, find_packages
 from distutils.command import build
 
 
@@ -57,7 +57,7 @@ def run(self, *args, **kwargs):
 
 setup(
     name='azure-functions-durable',
-    packages=find_packages(exclude=("tests","samples")),
+    packages=find_packages(exclude=("tests", "samples")),
     version='1.0.1ab',
     description='Durable Functions Support For Python Functionapp',
     license='MIT',

tests/fixtures.py → tests/conftest.py

@@ -1,5 +1,6 @@
 import pytest
-from azure.durable_functions.models.DurableOrchestrationBindings import DurableOrchestrationBindings
+from azure.durable_functions.models.DurableOrchestrationBindings import \
+    DurableOrchestrationBindings
 
 
 TASK_HUB_NAME = "DurableFunctionsHub"
@@ -10,18 +11,20 @@
 def get_binding_string():
     binding_string = '{"taskHubName":"TASK_HUB_NAME","creationUrls":{' \
                      '"createNewInstancePostUri":"BASE_URL/orchestrators/{functionName}[/{' \
-                     'instanceId}]?code=AUTH_CODE","createAndWaitOnNewInstancePostUri":"BASE_URL/orchestrators/{' \
-                     'functionName}[/{instanceId}]?timeout={timeoutInSeconds}&pollingInterval={' \
-                     'intervalInSeconds}&code=AUTH_CODE"},"managementUrls":{"id":"INSTANCEID",' \
-                     '"statusQueryGetUri":"BASE_URL/instances/INSTANCEID?taskHub=TASK_HUB_NAME&connection' \
-                     '=Storage&code=AUTH_CODE","sendEventPostUri":"BASE_URL/instances/INSTANCEID/raiseEvent/{' \
-                     'eventName}?taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE",' \
-                     '"terminatePostUri":"BASE_URL/instances/INSTANCEID/terminate?reason={' \
-                     'text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE",' \
+                     'instanceId}]?code=AUTH_CODE","createAndWaitOnNewInstancePostUri":' \
+                     '"BASE_URL/orchestrators/{functionName}[/{instanceId}]?' \
+                     'timeout={timeoutInSeconds}&pollingInterval={intervalInSeconds}' \
+                     '&code=AUTH_CODE"},"managementUrls":{"id":"INSTANCEID",' \
+                     '"statusQueryGetUri":"BASE_URL/instances/INSTANCEID?taskHub=' \
+                     'TASK_HUB_NAME&connection=Storage&code=AUTH_CODE","sendEventPostUri":' \
+                     '"BASE_URL/instances/INSTANCEID/raiseEvent/{eventName}?' \
+                     'taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE",' \
+                     '"terminatePostUri":"BASE_URL/instances/INSTANCEID/terminate?' \
+                     'reason={text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE",' \
                      '"rewindPostUri":"BASE_URL/instances/INSTANCEID/rewind?reason={' \
                      'text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE",' \
-                     '"purgeHistoryDeleteUri":"BASE_URL/instances/INSTANCEID?taskHub=TASK_HUB_NAME&connection' \
-                     '=Storage&code=AUTH_CODE"}}'
+                     '"purgeHistoryDeleteUri":"BASE_URL/instances/INSTANCEID?' \
+                     'taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE"}}'
 
     binding_string = replace_stand_in_bits(binding_string)
     return binding_string

tests/models/test_DurableOrchestrationBindings.py

@@ -1,46 +1,51 @@
-from tests.fixtures import *
+from tests.conftest import replace_stand_in_bits, TASK_HUB_NAME
 
 
 def test_extracts_task_hub_name(binding_info):
     assert TASK_HUB_NAME == binding_info.task_hub_name
 
 
 def test_extracts_create_new_instance_post_uri(binding_info):
-    expected_url = replace_stand_in_bits("BASE_URL/orchestrators/{functionName}[/{instanceId}]?code=AUTH_CODE")
+    expected_url = replace_stand_in_bits("BASE_URL/orchestrators/{functionName}[/{instanceId}]"
+                                         "?code=AUTH_CODE")
     assert expected_url == binding_info.creation_urls["createNewInstancePostUri"]
 
 
 def test_extracts_create_and_wait_on_new_instance_post_uri(binding_info):
-    expected_url = replace_stand_in_bits("BASE_URL/orchestrators/{functionName}[/{instanceId}]?timeout={"
-                                         "timeoutInSeconds}&pollingInterval={intervalInSeconds}&code=AUTH_CODE")
+    expected_url = replace_stand_in_bits("BASE_URL/orchestrators/{functionName}[/{instanceId}]?"
+                                         "timeout={timeoutInSeconds}&pollingInterval="
+                                         "{intervalInSeconds}&code=AUTH_CODE")
     assert expected_url == binding_info.creation_urls["createAndWaitOnNewInstancePostUri"]
 
 
 def test_extracts_status_query_get_uri(binding_info):
-    expected_url = replace_stand_in_bits("BASE_URL/instances/INSTANCEID?taskHub=TASK_HUB_NAME&connection=Storage"
-                                         "&code=AUTH_CODE")
+    expected_url = replace_stand_in_bits("BASE_URL/instances/INSTANCEID?taskHub=TASK_HUB_NAME"
+                                         "&connection=Storage&code=AUTH_CODE")
     assert expected_url == binding_info.management_urls["statusQueryGetUri"]
 
 
 def test_extracts_send_event_post_uri(binding_info):
     expected_url = replace_stand_in_bits("BASE_URL/instances/INSTANCEID/raiseEvent/{"
-                                         "eventName}?taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE")
+                                         "eventName}?taskHub=TASK_HUB_NAME&connection=Storage"
+                                         "&code=AUTH_CODE")
     assert expected_url == binding_info.management_urls["sendEventPostUri"]
 
 
 def test_extracts_terminate_post_uri(binding_info):
     expected_url = replace_stand_in_bits("BASE_URL/instances/INSTANCEID/terminate?reason={"
-                                         "text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE")
+                                         "text}&taskHub=TASK_HUB_NAME&connection=Storage"
+                                         "&code=AUTH_CODE")
     assert expected_url == binding_info.management_urls["terminatePostUri"]
 
 
 def test_extracts_rewind_post_uri(binding_info):
     expected_url = replace_stand_in_bits("BASE_URL/instances/INSTANCEID/rewind?reason={"
-                                         "text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE")
+                                         "text}&taskHub=TASK_HUB_NAME&connection=Storage"
+                                         "&code=AUTH_CODE")
     assert expected_url == binding_info.management_urls["rewindPostUri"]
 
 
 def test_extracts_purge_history_delete_uri(binding_info):
-    expected_url = replace_stand_in_bits("BASE_URL/instances/INSTANCEID?taskHub=TASK_HUB_NAME&connection=Storage&code"
-                                         "=AUTH_CODE")
+    expected_url = replace_stand_in_bits("BASE_URL/instances/INSTANCEID?"
+                                         "taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE")
     assert expected_url == binding_info.management_urls["purgeHistoryDeleteUri"]

tests/models/test_DurableOrchestrationClient.py

@@ -1,15 +1,16 @@
 import json
 
 from azure.durable_functions.models.DurableOrchestrationClient import DurableOrchestrationClient
-from tests.fixtures import *
+from tests.conftest import replace_stand_in_bits
 
 
 def test_get_start_new_url(binding_string):
     client = DurableOrchestrationClient(binding_string)
     instance_id = "abc123"
     function_name = "myfunction"
     start_new_url = client.get_start_new_url(instance_id, function_name)
-    expected_url = replace_stand_in_bits(f"BASE_URL/orchestrators/{function_name}/{instance_id}?code=AUTH_CODE")
+    expected_url = replace_stand_in_bits(
+        f"BASE_URL/orchestrators/{function_name}/{instance_id}?code=AUTH_CODE")
     assert expected_url == start_new_url