Acryl Executor is a Python library used internally by Acryl Agents and the DataHub ecosystem to define and execute modular tasks. It provides core abstractions like `BaseTask` and `Executor` for building extensible task-based workflows. As a sub-package of the larger DataHub project, its releases are often coupled with DataHub's evolution, though its own versioning is independent. The current version is 0.3.10, and it follows an active release cadence driven by DataHub development.
Install & Compatibility
Where this runs
tested against v0.3.15 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.910 runs
installs and imports cleanly · install 0.0s · import 0.000s · 201.6MB
glibcpy 3.10–3.910 runs
installs and imports cleanly · install 21.9s · import 0.000s · 201MB
208MB installed
● package 208MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
BaseTask
✓ from acryl.executor import BaseTask
✗ from acryl import BaseTask
Executor
✓ from acryl.executor import Executor
✗ from acryl import Executor
This quickstart demonstrates how to define a custom task by inheriting from `BaseTask`, implement its `execute` method, and then run it using the `Executor`. It shows how to pass configuration to the task and retrieve results. The example also illustrates how to safely access environment variables for sensitive data.
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}")
Audit
Dependencies
pydanticrequiredUsed for data validation and settings management within tasks.