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(mode: str, matrix: Dict[str, List[Any]])

Workflow context for action execution optimization.

Provides minimal context needed for actions to optimize across workflows. Actions receive specific data through inputs; context provides meta-information.

Attributes:

  • mode (str) –

    Execution mode ('dry-run', 'run', 'compare')

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

    Complete matrix definition for cross-job optimization

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(
    mode="run",
    matrix={"dataset": ["asia", "cancer"], "algorithm": ["pc", "ges"]},
)

# Context provides execution metadata for action optimization
print(f"Execution mode: {context.mode}")
print(f"Matrix definition: {context.matrix}")

# Actions can optimize across the full matrix space
if len(context.matrix.get("dataset", [])) > 1:
    print("Multi-dataset experiment - can pre-load data")

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 →