Refactor/task builder

pyzeebe/__init__.py

@@ -7,6 +7,7 @@
 from pyzeebe.job.job import Job
 from pyzeebe.job.job_status import JobStatus
 from pyzeebe.task.exception_handler import ExceptionHandler
-from pyzeebe.task.task_decorator import TaskDecorator
+from pyzeebe.task.task_config import TaskConfig, default_exception_handler
+from pyzeebe.task.types import TaskDecorator
 from pyzeebe.worker.task_router import ZeebeTaskRouter
 from pyzeebe.worker.worker import ZeebeWorker

pyzeebe/task/task.py

@@ -1,24 +1,16 @@
-from typing import Callable, List, Dict
+from dataclasses import dataclass
+from typing import Callable
 
-from pyzeebe.decorators.zeebe_decorator_base import ZeebeDecoratorBase
-from pyzeebe.job.job import Job
-from pyzeebe.task.exception_handler import ExceptionHandler
+from pyzeebe.task.task_config import TaskConfig
+from pyzeebe.task.types import JobHandler
 
 
-class Task(ZeebeDecoratorBase):
-    def __init__(self, task_type: str, task_handler: Callable[..., Dict], exception_handler: ExceptionHandler,
-                 timeout: int = 10000, max_jobs_to_activate: int = 32, variables_to_fetch: List[str] = None,
-                 before: List = None, after: List = None):
-        super().__init__(before=before, after=after)
+@dataclass
+class Task:
+    original_function: Callable
+    job_handler: JobHandler
+    config: TaskConfig
 
-        self.type = task_type
-        self.inner_function = task_handler
-        self.exception_handler = exception_handler
-        self.timeout = timeout
-        self.max_jobs_to_activate = max_jobs_to_activate
-        self.variables_to_fetch = variables_to_fetch or []
-        self.handler: Callable[[Job], Job] = None
-
-    def __repr__(self) -> str:
-        return str({"type": self.type, "timeout": self.timeout, "max_jobs_to_activate": self.max_jobs_to_activate,
-                    "variables_to_fetch": self.variables_to_fetch})
+    @property
+    def type(self):
+        return self.config.type

pyzeebe/task/task_builder.py

@@ -0,0 +1,76 @@
+import inspect
+import logging
+from typing import List, Callable, Dict, Tuple
+
+from pyzeebe import Job, TaskDecorator
+from pyzeebe.task.task import Task
+from pyzeebe.task.task_config import TaskConfig
+from pyzeebe.task.types import DecoratorRunner, JobHandler
+
+logger = logging.getLogger(__name__)
+
+
+def build_task(task_function: Callable, task_config: TaskConfig) -> Task:
+    if not task_config.variables_to_fetch:
+        task_config.variables_to_fetch = get_parameters_from_function(task_function)
+
+    if task_config.single_value:
+        task_function = convert_to_dict_function(task_function, task_config.variable_name)
+
+    return Task(task_function, build_job_handler(task_function, task_config), task_config)
+
+
+def build_job_handler(task_function: Callable, task_config: TaskConfig) -> JobHandler:
+    before_decorator_runner = create_decorator_runner(task_config.before)
+    after_decorator_runner = create_decorator_runner(task_config.after)
+
+    def job_handler(job: Job) -> Job:
+        job = before_decorator_runner(job)
+        job.variables, succeeded = run_original_task_function(task_function, task_config, job)
+        job = after_decorator_runner(job)
+        if succeeded:
+            job.set_success_status()
+        return job
+
+    return job_handler
+
+
+def run_original_task_function(task_function: Callable, task_config: TaskConfig, job: Job) -> Tuple[Dict, bool]:
+    try:
+        return task_function(**job.variables), True
+    except Exception as e:
+        logger.debug(f"Failed job: {job}. Error: {e}.")
+        task_config.exception_handler(e, job)
+        return job.variables, False
+
+
+def create_decorator_runner(decorators: List[TaskDecorator]) -> DecoratorRunner:
+    def decorator_runner(job: Job):
+        for decorator in decorators:
+            job = run_decorator(decorator, job)
+        return job
+
+    return decorator_runner
+
+
+def run_decorator(decorator: TaskDecorator, job: Job) -> Job:
+    try:
+        return decorator(job)
+    except Exception as e:
+        logger.warning(f"Failed to run decorator {decorator}. Exception: {e}")
+        return job
+
+
+def convert_to_dict_function(single_value_function: Callable, variable_name: str) -> Callable[..., Dict]:
+    def inner_fn(*args, **kwargs):
+        return {variable_name: single_value_function(*args, **kwargs)}
+
+    return inner_fn
+
+
+def get_parameters_from_function(fn: Callable) -> List[str]:
+    function_signature = inspect.signature(fn)
+    for parameter_name, parameter in function_signature.parameters.items():
+        if parameter.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
+            return []
+    return list(function_signature.parameters)

pyzeebe/task/task_config.py

@@ -0,0 +1,32 @@
+import logging
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+from pyzeebe.exceptions import NoVariableNameGiven
+from pyzeebe.job.job import Job
+from pyzeebe.task.exception_handler import ExceptionHandler
+from pyzeebe.task.types import TaskDecorator
+
+logger = logging.getLogger(__name__)
+
+
+def default_exception_handler(e: Exception, job: Job) -> None:
+    logger.warning(f"Task type: {job.type} - failed job {job}. Error: {e}.")
+    job.set_failure_status(f"Failed job. Error: {e}")
+
+
+@dataclass
+class TaskConfig:
+    type: str
+    exception_handler: ExceptionHandler = default_exception_handler
+    timeout: int = 10000
+    max_jobs_to_activate: int = 32
+    variables_to_fetch: Optional[List[str]] = None
+    single_value: bool = False
+    variable_name: Optional[str] = None
+    before: List[TaskDecorator] = field(default_factory=list)
+    after: List[TaskDecorator] = field(default_factory=list)
+
+    def __post_init__(self):
+        if self.single_value and not self.variable_name:
+            raise NoVariableNameGiven(self.type)

pyzeebe/task/types.py

@@ -0,0 +1,7 @@
+from typing import Callable
+
+from pyzeebe import Job
+
+DecoratorRunner = Callable[[Job], Job]
+JobHandler = Callable[[Job], Job]
+TaskDecorator = Callable[[Job], Job]