Skip to content

Action Registry

The action registry provides centralized discovery, management, and execution of workflow actions with plugin architecture support.

Core Classes

causaliq_workflow.registry

ActionRegistry 

ActionRegistry()

Registry for discovering and executing workflow actions dynamically.

Uses import-time introspection to automatically discover actions when packages are imported. No configuration needed - just import the package and use 'uses: package-name' in workflows.

Convention: Action packages should export an Action subclass named 'CausalIQAction' in their init.py file to avoid namespace collisions.

Attributes:

  • _instance (Optional[ActionRegistry]) –

    Singleton instance of the ActionRegistry

  • _actions (Dict[str, Type[Action]]) –

    Dictionary mapping action names to Action classes

  • _discovery_errors (List[str]) –

    List of errors encountered during action discovery

Initializes

_actions: Dictionary mapping action names to Action classes _discovery_errors: List to collect any discovery errors

Methods:

get_available_actions 

get_available_actions() -> Dict[str, Type[Action]]

Get dictionary of available action names to classes.

Returns:

  • Dict[str, Type[Action]] –

    Dictionary mapping action names to Action classes

get_action_class 

get_action_class(name: str) -> Type[Action]

Get action class by name.

Parameters:

  • name
    (str) –

    Action name

Returns:

  • Type[Action] –

    Action class

Raises:

has_action 

has_action(name: str) -> bool

Check if action is available.

Parameters:

  • name
    (str) –

    Action name to check

Returns:

  • bool –

    True if action is available

execute_action 

execute_action(
    name: str, inputs: Dict[str, Any], context: WorkflowContext
) -> Dict[str, Any]

Execute action with inputs and workflow context.

Parameters:

  • name
    (str) –

    Action name

  • inputs
    (Dict[str, Any]) –

    Action input parameters

  • context
    (WorkflowContext) –

    Complete workflow context

Returns:

  • Dict[str, Any] –

    Action outputs dictionary

Raises:

list_actions_by_package 

list_actions_by_package() -> Dict[str, List[str]]

Group actions by source package for documentation.

Returns:

  • Dict[str, List[str]] –

    Dictionary mapping package names to action lists

WorkflowContext dataclass 

WorkflowContext(
    workflow_id: str,
    matrix: Dict[str, List[Any]],
    current_job: Dict[str, Any],
    data_root: str,
    output_root: str,
    mode: str,
    all_jobs: Optional[List[Dict[str, Any]]] = None,
)

Complete workflow context passed to actions for optimization.

Attributes:

  • workflow_id (str) –

    Unique identifier for the workflow execution

  • matrix (Dict[str, List[Any]]) –

    Dictionary of matrix variables and their possible values

  • current_job (Dict[str, Any]) –

    Current matrix job parameters being executed

  • data_root (str) –

    Base path for data files in the workflow

  • output_root (str) –

    Base path for output files in the workflow

  • mode (str) –

    Execution mode ('run', 'validate', etc.)

  • all_jobs (Optional[List[Dict[str, Any]]]) –

    Complete list of all matrix jobs in the workflow

Exception Handling

ActionRegistryError

Raised when action registry operations fail.

This exception is raised when: - Requested action is not found in the registry - Action discovery fails during module scanning - Action validation fails - Other registry-related errors occur

Usage Examples

Basic Registry Operations

from causaliq_workflow.registry import ActionRegistry, ActionRegistryError

# Create registry instance
registry = ActionRegistry()

# List all available actions
actions = registry.get_available_actions()
for action_name, action_class in actions.items():
    print(f"Action: {action_name} (v{action_class.version})")

# Check if specific action exists
if registry.has_action("my-structure-learner"):
    action_class = registry.get_action_class("my-structure-learner")
    print(f"Found action: {action_class.description}")

# Execute action directly through registry
try:
    result = registry.execute_action(
        "my-structure-learner",
        {"data_path": "/data/asia.csv", "alpha": 0.05}
    )
    print(f"Execution result: {result}")
except ActionRegistryError as e:
    print(f"Registry error: {e}")

Plugin Discovery

# List actions by package (useful for plugin systems)
actions_by_package = registry.list_actions_by_package()
for package, actions in actions_by_package.items():
    print(f"Package: {package}")
    for action in actions:
        print(f"  - {action}")

# Discover actions from specific packages
registry = ActionRegistry(packages=["my_custom_actions"])
custom_actions = registry.get_available_actions()

Workflow Context

from causaliq_workflow.registry import WorkflowContext

# Create workflow context for action execution
context = WorkflowContext(
    workflow_id="experiment-001",
    step_name="structure-learning",
    matrix_vars={"dataset": "asia", "algorithm": "pc"},
    workflow_vars={"data_root": "/experiments/data"}
)

# Context provides complete workflow information to actions
print(f"Current step: {context.step_name}")
print(f"Matrix variables: {context.matrix_vars}")
print(f"Workflow root: {context.workflow_vars['data_root']}")

Architecture Notes

The ActionRegistry uses Python's module discovery system to automatically find and register actions. Actions are discovered by:

  1. Package scanning - Searches specified packages for Action subclasses
  2. Automatic registration - Actions register themselves via class definition
  3. Name-based lookup - Actions identified by their name class attribute
  4. Version tracking - Support for action versioning and compatibility

This design enables a flexible plugin architecture where actions can be distributed as separate packages and automatically discovered at runtime.

← Previous: Actions | Back to API Overview | Next: Workflow Engine →