Merge pull request #418 from botify-labs/bug/DATA-17956/Simpleflow-handle-failures-in-format-encode-format-decode

Author: ybastide

.pre-commit-config.yaml

@@ -12,13 +12,13 @@ repos:
       - id: check-added-large-files
 
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: 'v0.0.254'
+    rev: 'v0.0.278'
     hooks:
       - id: ruff
         args: [ --fix, --exit-non-zero-on-fix ]
 
   - repo: https://github.com/psf/black
-    rev: 23.1.0
+    rev: 23.7.0
     hooks:
       - id: black
         args: [--target-version=py37]

docs/src/features/error_handling.md

@@ -24,9 +24,12 @@ The `TaskFailureContext` currently have these members:
 
 * a_task: failed `ActivityTask` or `WorkflowTask`
 * task_name: activity or workflow name
-* exception: raised exception (shortcut to future.exception)
+* exception_class: `TaskException` or `WorkflowException`
+* exception: raised exception (shortcut to future.exception); None for unfinished tasks
 * retry_count: current retry count (0 for first retry)
 * task_error: for a TaskFailed exception, name of the inner exception if available
+* reason: `TaskFailed.reason` or `str(exception)`
+* details: `TaskFailed.details` or None
 * future: failed future
 * event: quite opaque dict, experimental
 * history: History object, experimental
@@ -38,7 +41,7 @@ The `TaskFailureContext.decision` (nothing in common with SWF’s decisions) sho
 * none: default value, continue with default handling (potential retries then abort)
 * abort: use default abort strategy
 * ignore: discard the exception, consider the task finished; the future’s result may be user-modified
-* cancel: mark the future as cancelled
+* cancel: mark the future as canceled
 * retry: schedule the task again; its args and kwargs may be user-modified (well, the whole task)
 * retry_later: schedule the task after retry_wait_timeout seconds, with args and kwargs potentially altered
 * handled: `on_task_failure` somewhat handled the failure; use the future and task it has possibly modified (one

examples/canvas.py

@@ -82,7 +82,7 @@ def run(self):
             )
         )
 
-        assert [None, 2, 6] == future.result, "Unexpected result {!r}".format(future.result)
+        assert [None, 2, 6] == future.result, f"Unexpected result {future.result!r}"
         print(f"Chain ignoring failure: {future.result}")
 
         # Failing inside a chain by default don't stop an upper chain
@@ -97,7 +97,7 @@ def run(self):
             )
         )
 
-        assert [[None], 2, 6] == future.result, "Unexpected result {!r}".format(future.result)
+        assert [[None], 2, 6] == future.result, f"Unexpected result {future.result!r}"
         print(f"Chain with failure in subchain: {future.result}")
 
         # But it can, too
@@ -116,7 +116,7 @@ def run(self):
             )
         )
 
-        assert [[[None]], 6] == future.result, "Unexpected result {!r}".format(future.result)
+        assert [[[None]], 6] == future.result, f"Unexpected result {future.result!r}"
         print(f"Chain with failure in sub-subchain: {future.result}")
 
         print("Finished!")

examples/child_workflow.py

@@ -62,7 +62,7 @@ def get_workflow_id(cls, *args, **kwargs):
         return kwargs.get("my_id")
 
     def run(self, my_id=None):
-        print("ChildWorkflowWithGetId: id={}, workflow_id={}".format(my_id, self.get_run_context().get("workflow_id")))
+        print(f"ChildWorkflowWithGetId: id={my_id}, workflow_id={self.get_run_context().get('workflow_id')}")
 
 
 class ParentWorkflow(Workflow):

script/release

@@ -55,7 +55,7 @@ def execute(command: list[str], ignore: bool = False, log: bool = False, dry_run
     :return: command output
     """
     if log or dry_run:
-        print("{}execute: {}".format("would " if dry_run else "", " ".join(quote(c) for c in command)))
+        print(f"{'would ' if dry_run else ''}execute: {' '.join(quote(c) for c in command)}")
     if dry_run:
         return ""
     env = os.environ.copy()
@@ -254,7 +254,7 @@ def input_new_version(current: Version) -> Version:
             try:
                 new_version = Version(new_version_str)
             except InvalidVersion as ex:
-                print("{}; Should be PEP 440-compatible (for instance in the form: 1.2.3)".format(ex))
+                print(f"{ex}; Should be PEP 440-compatible (for instance in the form: 1.2.3)")
         else:
             new_version = default_new_version
     return new_version
@@ -325,10 +325,11 @@ def main():
     release_tag(new_version, changes, dry_run)
 
     # push package to pypi
-    step("Generate and push package to {}".format(args.repository or args.repository_url or "pypi"))
-    execute(["python", "-m", "build", "--wheel"], log=True)
+    step(f"Generate and push package to {args.repository or args.repository_url or 'pypi'}")
+    execute(["python", "-m", "build"], log=True)
     wheel = f"dist/simpleflow-{new_version}-py3-none-any.whl"
-    cmd = ["twine", "upload", wheel]
+    tar_gz = f"dist/simpleflow-{new_version}.tar.gz"
+    cmd = ["twine", "upload", wheel, tar_gz]
     if args.repository:
         cmd += ["--repository", args.repository]
     elif args.repository_url:

simpleflow/__init__.py

@@ -5,6 +5,6 @@
 from .signal import WaitForSignal  # NOQA
 from .workflow import Workflow  # NOQA
 
-__version__ = "0.32.0"
+__version__ = "0.32.1dev1"
 __author__ = "Greg Leclercq"
 __license__ = "MIT"

simpleflow/activity.py

@@ -138,7 +138,7 @@ def name(self):
         return ".".join([prefix, name])
 
     def __repr__(self):
-        return "Activity(name={}, version={}, task_list={})".format(self.name, self.version, self.task_list)
+        return f"Activity(name={self.name}, version={self.version}, task_list={self.task_list})"
 
     def propagate_attribute(self, attr, val):
         setattr(self, attr, val)

simpleflow/canvas.py

@@ -107,13 +107,9 @@ def count_finished_activities(self):
         return sum(1 if a.finished else 0 for a in self.futures)
 
     def __repr__(self):
-        return "<{} at {:#x}, state={state}, exception={exception}, activities={activities}, futures={futures}>".format(
-            self.__class__.__name__,
-            id(self),
-            state=self._state,
-            exception=self._exception,
-            activities=self.activities,
-            futures=self.futures,
+        return (
+            f"<{self.__class__.__name__} at {id(self):#x}, state={self._state}, exception={self._exception},"
+            f" activities={self.activities}, futures={self.futures}>"
         )
 
 
@@ -185,7 +181,7 @@ def submit(self, executor):
         )
 
     def __repr__(self):
-        return "<{} at {:#x}, activities={!r}>".format(self.__class__.__name__, id(self), self.activities)
+        return f"<{self.__class__.__name__} at {id(self):#x}, activities={self.activities!r}>"
 
     def propagate_attribute(self, attr, val):
         """

simpleflow/command.py

@@ -206,12 +206,7 @@ def start_workflow(
         tag_list=tags,
         decision_tasks_timeout=decision_tasks_timeout,
     )
-    print(
-        "{workflow_id} {run_id}".format(
-            workflow_id=execution.workflow_id,
-            run_id=execution.run_id,
-        )
-    )
+    print(f"{execution.workflow_id} {execution.run_id}")
     return execution
 
 
@@ -252,12 +247,7 @@ def restart_workflow(domain, workflow_id, run_id):
         tag_list=ex.tag_list,
         decision_tasks_timeout=ex.decision_tasks_timeout,
     )
-    print(
-        "{workflow_id} {run_id}".format(
-            workflow_id=new_ex.workflow_id,
-            run_id=new_ex.run_id,
-        )
-    )
+    print(f"{new_ex.workflow_id} {new_ex.run_id}")
 
 
 def with_format(ctx):

simpleflow/download.py

@@ -67,7 +67,7 @@ def _check_binary_present(self):
         return os.access(self.local_location, os.X_OK)
 
     def _download_binary(self):
-        logger.info("Downloading binary: {} -> {}".format(self.remote_location, self.local_location))
+        logger.info(f"Downloading binary: {self.remote_location} -> {self.local_location}")
         bucket, path = self.remote_location.replace("s3://", "", 1).split("/", 1)
         # with FileLock(dest):
         pull(bucket, path, self.local_location)