[python client]: missing 1 required positional argument: 'task'

[pjanzen]

Hi All,

Following this readme I created a python worker.

simple_python_worker.py

from conductor.client.http.models import Task, TaskResult
from conductor.client.http.models.task_result_status import TaskResultStatus
from conductor.client.worker.worker_interface import WorkerInterface


class SimplePythonWorker(WorkerInterface):
    def execute(self, task: Task) -> TaskResult:
        print(task.input_data)
        task_result = self.get_task_result_from_task(task)
        task_result.add_output_data('worker_style', 'class')
        task_result.add_output_data('secret_number', 1234)
        task_result.add_output_data('is_it_true', False)
        task_result.status = TaskResultStatus.COMPLETED
        print(task_result)
        return task_result

    def get_polling_interval_in_seconds(self) -> float:
        # poll every 500ms
        return 1.0

client_worker.py

from conductor.client.automator.task_handler import TaskHandler
from conductor.client.configuration.configuration import Configuration
from conductor.client.worker.worker import Worker

from simple_python_worker import SimplePythonWorker


def main():
    configuration = Configuration(base_url='http://10.40.2.2:8080/')

    workers = [
        SimplePythonWorker(
            task_definition_name='python_task_example'
        ),
        Worker(
            task_definition_name='python_task_example',
            execute_function=SimplePythonWorker.execute,
            poll_interval=1,
        )
    ]

    with TaskHandler(workers, configuration) as task_handler:
        task_handler.start_processes()
        task_handler.join_processes()


if __name__ == '__main__':
    main()

The above code executes generated tasks but it also fails tasks.

{'options': ['-d', 1], 'emailAddress': 'sfqlttge@example.com', 'script': '/usr/local/bin/bla.sh'}
{'callback_after_seconds': None,
 'external_output_payload_storage_path': None,
 'logs': None,
 'output_data': {'is_it_true': False,
                 'secret_number': 1234,
                 'worker_style': 'class'},
 'reason_for_incompletion': None,
 'status': <TaskResultStatus.COMPLETED: 'COMPLETED'>,
 'sub_workflow_id': None,
 'task_id': '926bdc0a-24d8-4712-8b7d-2fc4f5e16971',
 'worker_id': 'airflow-node1',
 'workflow_instance_id': '81187582-baed-4b35-a96b-e0af8271cdf2'}

2022-08-09 05:34:54,590 [79903] conductor.client.automator.task_runner INFO     Failed to execute task, id: 10aa8ff8-324b-4189-a0d5-c6a0dafa00f6, workflow_instance_id: b4047f5b-aece-4e8b-805f-815dbb891946, task_definition_name: python_task_example, reason: Traceback (most recent call last):
  File "/opt/virtualenvs/conductor/lib/python3.8/site-packages/conductor/client/automator/task_runner.py", line 103, in __execute_task
    task_result = self.worker.execute(task)
  File "/opt/virtualenvs/conductor/lib/python3.8/site-packages/conductor/client/worker/worker.py", line 43, in execute
    return self.execute_function(task)
TypeError: execute() missing 1 required positional argument: 'task'

I am at a loss why this happens. Anybody here that have some pointers of a solution?

Regards,
Paul.

[pjanzen]

The example is not very clear on this but I removed the second Worker() from the workers list and the error was gone.

[apanicker-nflx]

This python worker is deprecate and not being maintained. Please use the python worker available here for an updated integration.

[pjanzen]

Hi,

That is the sdk that I use.

Regards,
Paul.

[gardusig]

Hey @pjanzen, we're working on some internal changes for the Python SDK that will address your issue. Stay tuned :)

I think there might be better ways of using workers btw, you have two options:

• Write a class and override methods as you wish (e.g. SimplePythonWorker)
• Just write a function and create a worker from it (e.g. Worker)

Mixing both approaches is counter intuitive IMO. What is the point of having SimplePythonWorker and reuse its execute method inside another worker? Classes exists for that purpose, you define it only once and instantiate it when needed.

Next release should avoid that but for now, if you want multiple workers of the same type, you can just replicate them by actually creating a new object for each. Example:

workers = [
   SimplePythonWorker(
     task_definition_name='python_task_example'
   ),
   SimplePythonWorker(
     task_definition_name='python_task_example'
   ),
   SimplePythonWorker(
     task_definition_name='python_task_example'
   ),
]

When you create a worker from a function while using a method from another class reference (e.g. SimplePythonWorker.execute) things can get messy, because you have two distinct workers sharing the same memory reference for the execute function, which is probably not the goal. What do you think about reviewing why you're doing this, then we can find the best way to achieve your goal?