Skip to content

Commit 022b514

Browse files
scgbearscgbear
authored
Merge pull request #6 from scgbear/refactor/first-pass
Refactor/Baseline Unit Tests
2 parents 5a6254e + 094361c commit 022b514

37 files changed

+2651
-297
lines changed

‎.gitignore‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -121,5 +121,14 @@ appsettings.*.json
121121
*.cer
122122
*.pfx
123123

124-
125-
124+
# pycharm
125+
.idea
126+
.root-refactor
127+
.sample-refactor
128+
.abstract-yield
129+
.bin
130+
131+
/samples/python_abstract_yield/bin/
132+
/samples/python_durable_bindings/bin/
133+
/samples/python_durable_bindings/.vs/
134+
/samples/python_durable_bindings/obj/

‎azure/durable_functions/__init__.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
from .orchestrator import Orchestrator
2+
from .models.DurableOrchestrationClient import DurableOrchestrationClient
23

34
__all__ = [
4-
'Orchestrator'
5+
'Orchestrator',
6+
'DurableOrchestrationClient'
57
]

‎azure/durable_functions/durable_orchestration_client.py‎

Lines changed: 0 additions & 7 deletions
This file was deleted.

‎azure/durable_functions/interfaces/ITaskMethods.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,5 +4,5 @@
44

55
class ITaskMethods:
66
def __init__(self):
7-
self.all: Callable[List[Task], TaskSet]
8-
self.any: Callable[List[Task], TaskSet]
7+
self.all: Callable[[List[Task]], TaskSet]
8+
self.any: Callable[[List[Task]], TaskSet]

‎azure/durable_functions/models/DurableOrchestrationBindings.py‎

Lines changed: 22 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,24 +1,29 @@
1+
import json
12
from typing import Dict
23

34

45
class DurableOrchestrationBindings:
5-
def __init__(self):
6-
self.taskHubName: str
7-
self.creationUrls: Dict[str, str]
8-
self.managementUrls: Dict[str, str]
6+
def __init__(self, client_data: str):
7+
context = json.loads(client_data)
8+
self.task_hub_name: str = context.get('taskHubName')
9+
self.creation_urls: Dict[str, str] = context.get('creationUrls')
10+
self.management_urls: Dict[str, str] = context.get('managementUrls')
11+
912

1013
'''
11-
"taskHubName":"DurableFunctionsHub",
12-
"creationUrls":{
13-
"createNewInstancePostUri":"http://localhost:7071/runtime/webhooks/durabletask/orchestrators/{functionName}[/{instanceId}]?code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
14-
"createAndWaitOnNewInstancePostUri":"http://localhost:7071/runtime/webhooks/durabletask/orchestrators/{functionName}[/{instanceId}]?timeout={timeoutInSeconds}&pollingInterval={intervalInSeconds}&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg=="
15-
},
16-
"managementUrls":{
17-
"id":"INSTANCEID",
18-
"statusQueryGetUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID?taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
19-
"sendEventPostUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/raiseEvent/{eventName}?taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
20-
"terminatePostUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/terminate?reason={text}&taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
21-
"rewindPostUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/rewind?reason={text}&taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
22-
"purgeHistoryDeleteUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID?taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg=="
14+
{
15+
"taskHubName":"DurableFunctionsHub",
16+
"creationUrls":{
17+
"createNewInstancePostUri":"http://localhost:7071/runtime/webhooks/durabletask/orchestrators/{functionName}[/{instanceId}]?code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
18+
"createAndWaitOnNewInstancePostUri":"http://localhost:7071/runtime/webhooks/durabletask/orchestrators/{functionName}[/{instanceId}]?timeout={timeoutInSeconds}&pollingInterval={intervalInSeconds}&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg=="
19+
},
20+
"managementUrls":{
21+
"id":"INSTANCEID",
22+
"statusQueryGetUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID?taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
23+
"sendEventPostUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/raiseEvent/{eventName}?taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
24+
"terminatePostUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/terminate?reason={text}&taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
25+
"rewindPostUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID/rewind?reason={text}&taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg==",
26+
"purgeHistoryDeleteUri":"http://localhost:7071/runtime/webhooks/durabletask/instances/INSTANCEID?taskHub=DurableFunctionsHub&connection=Storage&code=GBgDKQriGLABxpY/m5qcPd3R2sNafdRPE3/LcUSZEnuvOzTA1qD3Tg=="
27+
}
2328
}
24-
'''
29+
'''

‎azure/durable_functions/models/DurableOrchestrationClient.py‎

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,13 @@
1+
import requests
2+
import json
13
from typing import List
24

5+
from azure.durable_functions.models import DurableOrchestrationBindings
6+
37

48
class DurableOrchestrationClient:
59

6-
def __init__(self):
10+
def __init__(self, context: str):
711
self.taskHubName: str
812

913
self.uniqueWebhookOrigins: List[str]
@@ -21,3 +25,24 @@ def __init__(self):
2125
self._showHistoryQueryKey: str = "showHistory"
2226
self._showHistoryOutputQueryKey: str = "showHistoryOutput"
2327
self._showInputQueryKey: str = "showInput"
28+
self._orchestrationBindings: DurableOrchestrationBindings = DurableOrchestrationBindings(context)
29+
30+
def start_new(self,
31+
orchestration_function_name: str,
32+
instance_id: str,
33+
client_input):
34+
request_url = self.get_start_new_url(instance_id, orchestration_function_name)
35+
36+
result = requests.post(request_url, json=self.get_json_input(client_input))
37+
return result
38+
39+
@staticmethod
40+
def get_json_input(client_input):
41+
return json.dumps(client_input) if client_input is not None else None
42+
43+
def get_start_new_url(self, instance_id, orchestration_function_name):
44+
request_url = self._orchestrationBindings.creation_urls['createNewInstancePostUri']
45+
request_url = request_url.replace(self._functionNamePlaceholder, orchestration_function_name)
46+
request_url = request_url.replace(self._instanceIdPlaceholder,
47+
f'/{instance_id}' if instance_id is not None else '')
48+
return request_url

‎azure/durable_functions/models/DurableOrchestrationContext.py‎

Lines changed: 31 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,29 +1,42 @@
1-
from datetime import datetime
1+
import json
2+
import logging
3+
from typing import List, Any, Dict
4+
5+
from dateutil.parser import parse as dt_parse
6+
7+
from . import (RetryOptions)
8+
from .history import HistoryEvent, HistoryEventType
9+
from ..interfaces import IAction
210
from ..interfaces import ITaskMethods
3-
from . import (Task, RetryOptions)
11+
from ..models.Task import Task
12+
from ..tasks import call_activity, task_all
413

514

615
class DurableOrchestrationContext:
716

817
def __init__(self,
9-
instanceId,
10-
isReplaying,
11-
parentInstanceId,
12-
callActivity,
13-
task_all,
14-
currentUtcDateTime):
15-
self.instanceId: str = instanceId
16-
self.isReplaying: bool = isReplaying
17-
self.parentInstanceId: str = parentInstanceId
18-
self.callActivity = callActivity
19-
self.task_all = task_all
20-
self.currentUtcDateTime = currentUtcDateTime
21-
22-
# self.currentUtcDateTime: Date
23-
self.currentUtcDateTime: datetime
18+
context_string: str):
19+
context: Dict[str, Any] = json.loads(context_string)
20+
logging.warning(f"!!!Calling orchestrator handle {context}")
21+
self.histories: List[HistoryEvent] = context.get("history")
22+
self.instanceId = context.get("instanceId")
23+
self.isReplaying = context.get("isReplaying")
24+
self.parentInstanceId = context.get("parentInstanceId")
25+
self.callActivity = lambda n, i: call_activity(
26+
state=self.histories,
27+
name=n,
28+
input_=i)
29+
self.task_all = lambda t: task_all(state=self.histories, tasks=t)
30+
self.decision_started_event: HistoryEvent = list(filter(
31+
# HistoryEventType.OrchestratorStarted
32+
lambda e_: e_["EventType"] == HistoryEventType.OrchestratorStarted,
33+
self.histories))[0]
34+
self.currentUtcDateTime = dt_parse(self.decision_started_event["Timestamp"])
35+
self.newGuidCounter = 0
36+
self.actions: List[List[IAction]] = []
2437
self.Task: ITaskMethods
2538

26-
def callActivity(name: str, input=None) -> Task:
39+
def callActivity(name: str, input_=None) -> Task:
2740
raise NotImplementedError("This is a placeholder.")
2841

2942
def callActivityWithRetry(

‎azure/durable_functions/models/OrchestratorState.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,12 +29,14 @@ def to_json(self) -> Dict[str, Any]:
2929
action_dict['functionName'] = action_obj.functionName
3030
if hasattr(action_obj, 'input'):
3131
action_dict['input'] = action_obj.input
32-
3332
action_result_list.append(action_dict)
3433
json_dict['actions'].append(action_result_list)
35-
json_dict['output'] = self.output
36-
json_dict['error'] = self.error
37-
json_dict['customStatus'] = self.customStatus
34+
if self.output:
35+
json_dict['output'] = self.output
36+
if self.error:
37+
json_dict['error'] = self.error
38+
if self.customStatus:
39+
json_dict['customStatus'] = self.customStatus
3840
return json_dict
3941

4042
def to_json_string(self) -> str:

‎azure/durable_functions/models/Task.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@
33

44

55
class Task:
6+
action: IAction
7+
68
def __init__(self, isCompleted, isFaulted, action,
79
result=None, timestamp=None, id=None, exc=None):
810
self.isCompleted: bool = isCompleted

‎azure/durable_functions/models/history/HistoryEvent.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,5 +7,5 @@ def __init__(self):
77
self.EventType: HistoryEventType
88
self.EventId: int
99
self.IsPlayed: bool
10-
self.Timestamp: datetime
10+
self.Timestamp: str
1111
self.IsProcessed: bool = False

0 commit comments

Comments
 (0)