acryl-executor

import os from typing import Dict, Any, NamedTuple from abc import ABC, abstractmethod from acryl.executor.sdk.tasks.base_task import BaseTask, TaskResult from acryl.executor.sdk.executor import Executor class MySimpleTaskResult(NamedTuple): status: str message: str class MySimpleTask(BaseTask): def execute(self, config: Dict[str, Any]) -> MySimpleTaskResult: name = config.get('name', 'World') # Example of using an environment variable for configuration (e.g., for API keys) secret_key = os.environ.get('MY_SECRET_KEY', 'default_secret') print(f"Executing MySimpleTask for {name} with secret: {secret_key[:4]}...") return MySimpleTaskResult(status='SUCCESS', message=f'Hello, {name}!') # Configure with actual environment variable or a placeholder for quick run os.environ['MY_SECRET_KEY'] = os.environ.get('MY_SECRET_KEY', 'some_default_value_for_testing') # 1. Instantiate the Executor executor = Executor() # 2. Define a task instance my_task = MySimpleTask() # 3. Define task configuration task_config = {"name": "DataHub User"} # 4. Execute the task print("\n--- Executing MySimpleTask ---") result = executor.execute_task(my_task, task_config) print(f"Task Result: {result.status}, Message: {result.message}") # Example with different config task_config_2 = {"name": "Acryl Team"} print("\n--- Executing MySimpleTask with different config ---") result_2 = executor.execute_task(my_task, task_config_2) print(f"Task Result: {result_2.status}, Message: {result_2.message}")