Skip to content

API Reference

This section contains automatic API documentation for Automax.

Plugin System

automax.plugins.base

Base classes and interfaces for Automax plugins.

This module defines the abstract base class and metadata structure that all plugins must implement.

BasePlugin

Bases: ABC

Abstract base class defining the plugin interface.

All Automax plugins must inherit from this class and implement the required abstract methods.

Source code in src/automax/plugins/base.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class BasePlugin(ABC):
    """
    Abstract base class defining the plugin interface.

    All Automax plugins must inherit from this class and implement the required abstract
    methods.

    """

    METADATA = PluginMetadata(name="base")

    # Schema for parameter validation - can be overridden by subclasses
    SCHEMA = {}

    def __init__(self, config: Dict[str, Any]):
        """
        Initialize plugin with configuration.

        Args:
            config: Plugin configuration dictionary

        Raises:
            PluginConfigurationError: If required configuration is missing

        """
        self.config = config
        self.logger = self._setup_logger()
        self._validate_config()

    def _setup_logger(self) -> logging.Logger:
        """
        Configure standardized logger for plugin.

        Returns:
            Configured logger instance

        """
        logger_name = f"automax.plugins.{self.METADATA.name}"
        return logging.getLogger(logger_name)

    def _validate_config(self):
        """
        Validate plugin configuration against required fields.

        Raises:
            PluginConfigurationError: If required configuration keys are missing

        """
        missing_required = [
            key for key in self.METADATA.required_config if key not in self.config
        ]

        if missing_required:
            raise PluginConfigurationError(
                f"Missing required configuration: {missing_required}"
            )

    @abstractmethod
    def execute(self) -> Dict[str, Any]:
        """
        Execute the plugin's main functionality.

        Returns:
            Dictionary containing execution results

        """
        pass

    def validate(self) -> bool:
        """
        Validate plugin readiness before execution.

        Returns:
            True if plugin is ready for execution

        """
        return True

    def cleanup(self):
        """
        Release resources and perform cleanup operations.
        """
        pass

    def get_help(self) -> str:
        """
        Get usage documentation for the plugin.

        Returns:
            Help text describing plugin usage and configuration

        """
        return self.METADATA.description

    def get_example_config(self) -> Dict[str, Any]:
        """
        Generate example configuration for the plugin.

        Returns:
            Example configuration dictionary

        """
        example = {}
        for key in self.METADATA.required_config:
            example[key] = f"<{key}>"
        for key in self.METADATA.optional_config:
            example[key] = f"[{key}]"
        return example

__init__(config)

Initialize plugin with configuration.

Parameters:

Name Type Description Default
config Dict[str, Any]

Plugin configuration dictionary

 required

Raises:

Type Description
PluginConfigurationError

If required configuration is missing
Source code in src/automax/plugins/base.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(self, config: Dict[str, Any]):
    """
    Initialize plugin with configuration.

    Args:
        config: Plugin configuration dictionary

    Raises:
        PluginConfigurationError: If required configuration is missing

    """
    self.config = config
    self.logger = self._setup_logger()
    self._validate_config()

cleanup()

Release resources and perform cleanup operations.

Source code in src/automax/plugins/base.py 
131
132
133
134
135
def cleanup(self):
    """
    Release resources and perform cleanup operations.
    """
    pass

execute() abstractmethod

Execute the plugin's main functionality.

Returns:

Type Description
Dict[str, Any]

Dictionary containing execution results
Source code in src/automax/plugins/base.py 
110
111
112
113
114
115
116
117
118
119
@abstractmethod
def execute(self) -> Dict[str, Any]:
    """
    Execute the plugin's main functionality.

    Returns:
        Dictionary containing execution results

    """
    pass

get_example_config()

Generate example configuration for the plugin.

Returns:

Type Description
Dict[str, Any]

Example configuration dictionary
Source code in src/automax/plugins/base.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def get_example_config(self) -> Dict[str, Any]:
    """
    Generate example configuration for the plugin.

    Returns:
        Example configuration dictionary

    """
    example = {}
    for key in self.METADATA.required_config:
        example[key] = f"<{key}>"
    for key in self.METADATA.optional_config:
        example[key] = f"[{key}]"
    return example

get_help()

Get usage documentation for the plugin.

Returns:

Type Description
str

Help text describing plugin usage and configuration
Source code in src/automax/plugins/base.py 
137
138
139
140
141
142
143
144
145
def get_help(self) -> str:
    """
    Get usage documentation for the plugin.

    Returns:
        Help text describing plugin usage and configuration

    """
    return self.METADATA.description

validate()

Validate plugin readiness before execution.

Returns:

Type Description
bool

True if plugin is ready for execution
Source code in src/automax/plugins/base.py 
121
122
123
124
125
126
127
128
129
def validate(self) -> bool:
    """
    Validate plugin readiness before execution.

    Returns:
        True if plugin is ready for execution

    """
    return True

PluginMetadata

Standardized metadata for plugin registration and discovery.

Attributes:

Name Type Description
name

Unique identifier for the plugin
version

Semantic version of the plugin
description

Brief description of plugin functionality
author

Plugin author or maintainer
category

Functional category for grouping
tags

Keywords for search and filtering
required_config

List of required configuration keys
optional_config

List of optional configuration keys
Source code in src/automax/plugins/base.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class PluginMetadata:
    """
    Standardized metadata for plugin registration and discovery.

    Attributes:
        name: Unique identifier for the plugin
        version: Semantic version of the plugin
        description: Brief description of plugin functionality
        author: Plugin author or maintainer
        category: Functional category for grouping
        tags: Keywords for search and filtering
        required_config: List of required configuration keys
        optional_config: List of optional configuration keys

    """

    def __init__(
        self,
        name: str,
        version: str = "1.0.0",
        description: str = "",
        author: str = "Automax",
        category: str = "general",
        tags: List[str] = None,
        required_config: List[str] = None,
        optional_config: List[str] = None,
    ):
        self.name = name
        self.version = version
        self.description = description
        self.author = author
        self.category = category
        self.tags = tags or []
        self.required_config = required_config or []
        self.optional_config = optional_config or []

automax.plugins.exceptions

Custom exception hierarchy for plugin error handling.

This module defines standardized exceptions for plugin configuration, execution, and validation errors.

PluginConfigurationError

Bases: PluginError

Raised when plugin configuration is invalid or incomplete.

Source code in src/automax/plugins/exceptions.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class PluginConfigurationError(PluginError):
    """
    Raised when plugin configuration is invalid or incomplete.
    """

    def __init__(
        self,
        message: str,
        plugin_name: Optional[str] = None,
        config: Optional[Dict[str, Any]] = None,
        original_error: Optional[Exception] = None,
        missing_keys: Optional[List[str]] = None,
        invalid_keys: Optional[List[str]] = None,
    ):
        super().__init__(message, plugin_name, config, original_error)
        self.missing_keys = missing_keys or []
        self.invalid_keys = invalid_keys or []

    def __str__(self) -> str:
        """
        String representation with configuration details.
        """
        base_message = super().__str__()
        details = []
        if self.missing_keys:
            details.append(f"missing keys: {self.missing_keys}")
        if self.invalid_keys:
            details.append(f"invalid keys: {self.invalid_keys}")
        if details:
            return f"{base_message} ({', '.join(details)})"
        return base_message

__str__()

String representation with configuration details.

Source code in src/automax/plugins/exceptions.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
def __str__(self) -> str:
    """
    String representation with configuration details.
    """
    base_message = super().__str__()
    details = []
    if self.missing_keys:
        details.append(f"missing keys: {self.missing_keys}")
    if self.invalid_keys:
        details.append(f"invalid keys: {self.invalid_keys}")
    if details:
        return f"{base_message} ({', '.join(details)})"
    return base_message

PluginError

Bases: Exception

Base exception for all plugin-related errors.

Source code in src/automax/plugins/exceptions.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class PluginError(Exception):
    """
    Base exception for all plugin-related errors.
    """

    def __init__(
        self,
        message: str,
        plugin_name: Optional[str] = None,
        config: Optional[Dict[str, Any]] = None,
        original_error: Optional[Exception] = None,
    ):
        self.message = message
        self.plugin_name = plugin_name
        self.config = config or {}
        self.original_error = original_error
        super().__init__(self.message)

    def __str__(self) -> str:
        """
        String representation of the error.
        """
        base_message = f"PluginError: {self.message}"
        if self.plugin_name:
            base_message = f"[{self.plugin_name}] {base_message}"
        return base_message

    def to_dict(self) -> Dict[str, Any]:
        """
        Convert error to dictionary for serialization.

        Returns:
            Dictionary containing error details

        """
        return {
            "error_type": self.__class__.__name__,
            "message": self.message,
            "plugin_name": self.plugin_name,
            "config": {
                k: v
                for k, v in self.config.items()
                if not isinstance(v, (str, int, float, bool)) or len(str(v)) < 100
            },
        }

__str__()

String representation of the error.

Source code in src/automax/plugins/exceptions.py 
32
33
34
35
36
37
38
39
def __str__(self) -> str:
    """
    String representation of the error.
    """
    base_message = f"PluginError: {self.message}"
    if self.plugin_name:
        base_message = f"[{self.plugin_name}] {base_message}"
    return base_message

to_dict()

Convert error to dictionary for serialization.

Returns:

Type Description
Dict[str, Any]

Dictionary containing error details
Source code in src/automax/plugins/exceptions.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def to_dict(self) -> Dict[str, Any]:
    """
    Convert error to dictionary for serialization.

    Returns:
        Dictionary containing error details

    """
    return {
        "error_type": self.__class__.__name__,
        "message": self.message,
        "plugin_name": self.plugin_name,
        "config": {
            k: v
            for k, v in self.config.items()
            if not isinstance(v, (str, int, float, bool)) or len(str(v)) < 100
        },
    }

PluginExecutionError

Bases: PluginError

Raised when plugin execution fails.

Source code in src/automax/plugins/exceptions.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class PluginExecutionError(PluginError):
    """
    Raised when plugin execution fails.
    """

    def __init__(
        self,
        message: str,
        plugin_name: Optional[str] = None,
        config: Optional[Dict[str, Any]] = None,
        original_error: Optional[Exception] = None,
        execution_step: Optional[str] = None,
    ):
        super().__init__(message, plugin_name, config, original_error)
        self.execution_step = execution_step

    def __str__(self) -> str:
        """
        String representation with execution context.
        """
        base_message = super().__str__()
        if self.execution_step:
            return f"{base_message} (step: {self.execution_step})"
        return base_message

__str__()

String representation with execution context.

Source code in src/automax/plugins/exceptions.py 
110
111
112
113
114
115
116
117
def __str__(self) -> str:
    """
    String representation with execution context.
    """
    base_message = super().__str__()
    if self.execution_step:
        return f"{base_message} (step: {self.execution_step})"
    return base_message

PluginSecurityError

Bases: PluginError

Raised when security constraints are violated.

Source code in src/automax/plugins/exceptions.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class PluginSecurityError(PluginError):
    """
    Raised when security constraints are violated.
    """

    def __init__(
        self,
        message: str,
        plugin_name: Optional[str] = None,
        config: Optional[Dict[str, Any]] = None,
        original_error: Optional[Exception] = None,
        security_context: Optional[Dict[str, Any]] = None,
    ):
        super().__init__(message, plugin_name, config, original_error)
        self.security_context = security_context or {}

    def __str__(self) -> str:
        """
        String representation with security context.
        """
        base_message = super().__str__()
        if self.security_context:
            context_str = ", ".join(
                f"{k}={v}" for k, v in self.security_context.items()
            )
            return f"{base_message} (security context: {context_str})"
        return base_message

__str__()

String representation with security context.

Source code in src/automax/plugins/exceptions.py 
163
164
165
166
167
168
169
170
171
172
173
def __str__(self) -> str:
    """
    String representation with security context.
    """
    base_message = super().__str__()
    if self.security_context:
        context_str = ", ".join(
            f"{k}={v}" for k, v in self.security_context.items()
        )
        return f"{base_message} (security context: {context_str})"
    return base_message

PluginValidationError

Bases: PluginError

Raised when plugin validation fails.

Source code in src/automax/plugins/exceptions.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class PluginValidationError(PluginError):
    """
    Raised when plugin validation fails.
    """

    def __init__(
        self,
        message: str,
        plugin_name: Optional[str] = None,
        config: Optional[Dict[str, Any]] = None,
        original_error: Optional[Exception] = None,
        validation_errors: Optional[List[str]] = None,
    ):
        super().__init__(message, plugin_name, config, original_error)
        self.validation_errors = validation_errors or []

    def __str__(self) -> str:
        """
        String representation with validation details.
        """
        base_message = super().__str__()
        if self.validation_errors:
            errors = "; ".join(self.validation_errors)
            return f"{base_message} (validation errors: {errors})"
        return base_message

__str__()

String representation with validation details.

Source code in src/automax/plugins/exceptions.py 
136
137
138
139
140
141
142
143
144
def __str__(self) -> str:
    """
    String representation with validation details.
    """
    base_message = super().__str__()
    if self.validation_errors:
        errors = "; ".join(self.validation_errors)
        return f"{base_message} (validation errors: {errors})"
    return base_message

handle_plugin_errors(plugin_name)

Decorator for automatic plugin error handling.

Parameters:

Name Type Description Default
plugin_name str

Name of the plugin for error context

 required

Returns:

Type Description
Callable[[F], F]

Decorated function with error handling
Source code in src/automax/plugins/exceptions.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def handle_plugin_errors(plugin_name: str) -> Callable[[F], F]:
    """
    Decorator for automatic plugin error handling.

    Args:
        plugin_name: Name of the plugin for error context

    Returns:
        Decorated function with error handling

    """

    def decorator(func: F) -> F:
        def wrapper(self, *args, **kwargs):
            try:
                return func(self, *args, **kwargs)
            except PluginError:
                raise
            except Exception as e:
                raise PluginExecutionError(
                    message=f"Unexpected error: {str(e)}",
                    plugin_name=plugin_name,
                    config=getattr(self, "config", {}),
                    original_error=e,
                ) from e

        return wrapper

    return decorator

automax.plugins.registry

Plugin registry for dynamic discovery and management.

This module provides centralized plugin registration, discovery, and metadata management.

PluginRegistry

Central registry for plugin registration and discovery.

Source code in src/automax/plugins/registry.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class PluginRegistry:
    """
    Central registry for plugin registration and discovery.
    """

    def __init__(self):
        self._plugins: Dict[str, Type[BasePlugin]] = {}
        self._metadata: Dict[str, PluginMetadata] = {}
        self._loaded = False
        self.logger = logging.getLogger("automax.plugins.registry")

    def register(self, plugin_class: Type[BasePlugin]):
        """
        Register a plugin class in the registry.
        """
        metadata = plugin_class.METADATA
        self._validate_plugin_class(plugin_class, metadata)

        self._plugins[metadata.name] = plugin_class
        self._metadata[metadata.name] = metadata

        self.logger.debug(f"Registered plugin: {metadata.name} v{metadata.version}")

    def _validate_plugin_class(
        self, plugin_class: Type[BasePlugin], metadata: PluginMetadata
    ):
        """
        Validate plugin class before registration.
        """
        if not metadata.name or not isinstance(metadata.name, str):
            raise ValueError(f"Plugin name must be non-empty string: {plugin_class}")

        if (
            metadata.name in self._plugins
            and self._plugins[metadata.name] != plugin_class
        ):
            raise ValueError(f"Plugin name conflict: {metadata.name}")

        if not hasattr(plugin_class, "execute") or not callable(plugin_class.execute):
            raise ValueError(f"Plugin must implement execute method: {metadata.name}")

    def get_plugin_class(self, name: str) -> Type[BasePlugin]:
        """
        Retrieve plugin class by name.
        """
        if name not in self._plugins:
            raise KeyError(f"Plugin not found: {name}")
        return self._plugins[name]

    def get_metadata(self, name: str) -> PluginMetadata:
        """
        Retrieve plugin metadata by name.
        """
        if name not in self._metadata:
            raise KeyError(f"Plugin metadata not found: {name}")
        return self._metadata[name]

    def list_plugins(self, category: Optional[str] = None) -> List[str]:
        """
        List all registered plugins.
        """
        if category:
            return [
                name
                for name, metadata in self._metadata.items()
                if metadata.category == category
            ]
        return list(self._plugins.keys())

    def get_plugins_by_tag(self, tag: str) -> List[str]:
        """
        Find plugins by tag.
        """
        return [
            name for name, metadata in self._metadata.items() if tag in metadata.tags
        ]

    def load_all_plugins(self):
        """
        Discover and load all plugins from the plugins directory.
        """
        if self._loaded:
            return

        plugins_package = "automax.plugins"
        plugins_path = Path(__file__).parent

        for module_file in plugins_path.glob("*.py"):
            if module_file.name in [
                "__init__.py",
                "base.py",
                "exceptions.py",
                "registry.py",
            ]:
                continue

            module_name = module_file.stem
            try:
                self._load_plugin_module(plugins_package, module_name)
            except Exception as e:
                self.logger.warning(f"Failed to load plugin module {module_name}: {e}")

        self._loaded = True
        self.logger.info(f"Loaded {len(self._plugins)} plugins")

    def _load_plugin_module(self, package: str, module_name: str):
        """
        Load and register plugins from a module.
        """
        try:
            full_module_name = f"{package}.{module_name}"
            module = importlib.import_module(full_module_name)

            # Find all classes in the module that are BasePlugin subclasses
            for name, obj in inspect.getmembers(module, inspect.isclass):
                if (
                    obj.__module__ == module.__name__
                    and issubclass(obj, BasePlugin)
                    and obj is not BasePlugin
                ):
                    self.register(obj)
                    self.logger.debug(f"Discovered plugin: {obj.METADATA.name}")

        except Exception as e:
            self.logger.error(f"Error loading module {module_name}: {e}")
            raise

get_metadata(name)

Retrieve plugin metadata by name.

Source code in src/automax/plugins/registry.py 
67
68
69
70
71
72
73
def get_metadata(self, name: str) -> PluginMetadata:
    """
    Retrieve plugin metadata by name.
    """
    if name not in self._metadata:
        raise KeyError(f"Plugin metadata not found: {name}")
    return self._metadata[name]

get_plugin_class(name)

Retrieve plugin class by name.

Source code in src/automax/plugins/registry.py 
59
60
61
62
63
64
65
def get_plugin_class(self, name: str) -> Type[BasePlugin]:
    """
    Retrieve plugin class by name.
    """
    if name not in self._plugins:
        raise KeyError(f"Plugin not found: {name}")
    return self._plugins[name]

get_plugins_by_tag(tag)

Find plugins by tag.

Source code in src/automax/plugins/registry.py 
87
88
89
90
91
92
93
def get_plugins_by_tag(self, tag: str) -> List[str]:
    """
    Find plugins by tag.
    """
    return [
        name for name, metadata in self._metadata.items() if tag in metadata.tags
    ]

list_plugins(category=None)

List all registered plugins.

Source code in src/automax/plugins/registry.py 
75
76
77
78
79
80
81
82
83
84
85
def list_plugins(self, category: Optional[str] = None) -> List[str]:
    """
    List all registered plugins.
    """
    if category:
        return [
            name
            for name, metadata in self._metadata.items()
            if metadata.category == category
        ]
    return list(self._plugins.keys())

load_all_plugins()

Discover and load all plugins from the plugins directory.

Source code in src/automax/plugins/registry.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def load_all_plugins(self):
    """
    Discover and load all plugins from the plugins directory.
    """
    if self._loaded:
        return

    plugins_package = "automax.plugins"
    plugins_path = Path(__file__).parent

    for module_file in plugins_path.glob("*.py"):
        if module_file.name in [
            "__init__.py",
            "base.py",
            "exceptions.py",
            "registry.py",
        ]:
            continue

        module_name = module_file.stem
        try:
            self._load_plugin_module(plugins_package, module_name)
        except Exception as e:
            self.logger.warning(f"Failed to load plugin module {module_name}: {e}")

    self._loaded = True
    self.logger.info(f"Loaded {len(self._plugins)} plugins")

register(plugin_class)

Register a plugin class in the registry.

Source code in src/automax/plugins/registry.py 
29
30
31
32
33
34
35
36
37
38
39
def register(self, plugin_class: Type[BasePlugin]):
    """
    Register a plugin class in the registry.
    """
    metadata = plugin_class.METADATA
    self._validate_plugin_class(plugin_class, metadata)

    self._plugins[metadata.name] = plugin_class
    self._metadata[metadata.name] = metadata

    self.logger.debug(f"Registered plugin: {metadata.name} v{metadata.version}")

register_plugin(plugin_class)

Class decorator for automatic plugin registration.

Source code in src/automax/plugins/registry.py 
150
151
152
153
154
155
def register_plugin(plugin_class: Type[BasePlugin]) -> Type[BasePlugin]:
    """
    Class decorator for automatic plugin registration.
    """
    global_registry.register(plugin_class)
    return plugin_class

Core Managers

automax.core.managers.plugin_manager

Plugin Manager for Automax.

This module handles plugin loading, registration, and execution.

PluginManager

Manages plugin loading and execution.

This class provides backward compatibility with the existing system while integrating with the new plugin registry.

Source code in src/automax/core/managers/plugin_manager.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class PluginManager:
    """
    Manages plugin loading and execution.

    This class provides backward compatibility with the existing system while
    integrating with the new plugin registry.

    """

    def __init__(self, logger=None):
        """
        Initialize the PluginManager.

        Args:
            logger (optional): Logger instance for debug/info/error messages.

        """
        self.logger = logger
        self._legacy_plugins: Dict[str, Any] = {}
        self._load_legacy_plugins()

    def _load_legacy_plugins(self):
        """
        Load plugins using legacy method for backward compatibility.
        """
        import importlib
        from pathlib import Path

        if self.logger:
            self.logger.debug("Loading legacy plugins...")

        plugins_dir = Path(__file__).parent.parent.parent / "plugins"

        for module_file in plugins_dir.glob("*.py"):
            if module_file.name in [
                "__init__.py",
                "base.py",
                "exceptions.py",
                "registry.py",
            ]:
                continue

            module_name = module_file.stem
            try:
                module = importlib.import_module(f"automax.plugins.{module_name}")

                # Look for legacy plugin functions
                for attr_name in dir(module):
                    attr = getattr(module, attr_name)
                    if callable(attr) and not attr_name.startswith("_"):
                        self._legacy_plugins[module_name] = attr
                        break

            except Exception as e:
                self.logger.warning(f"Failed to load legacy plugin {module_name}: {e}")

    def execute_plugin(self, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Execute a plugin with given configuration.

        This method first tries the new registry system, then falls back to legacy
        plugins for backward compatibility.

        """
        if self.logger:
            self.logger.debug(f"Executing plugin {name} with configuration {config}")

        # Try new registry system first
        global_registry.load_all_plugins()

        if name in global_registry.list_plugins():
            try:
                plugin_class = global_registry.get_plugin_class(name)
                plugin_instance = plugin_class(config)
                return plugin_instance.execute()
            except Exception as e:
                raise PluginExecutionError(
                    f"Plugin {name} execution failed: {e}"
                ) from e

        # Fall back to legacy system
        elif name in self._legacy_plugins:
            self.logger.warning(f"Using legacy plugin: {name}")
            try:
                return self._legacy_plugins[name](config)
            except Exception as e:
                raise PluginExecutionError(
                    f"Legacy plugin {name} execution failed: {e}"
                ) from e

        else:
            available_plugins = global_registry.list_plugins() + list(
                self._legacy_plugins.keys()
            )
            raise PluginError(
                f"Plugin '{name}' not found. Available plugins: {available_plugins}"
            )

    def get_plugin(self, plugin_name: str):
        """
        Get plugin class by name.
        """
        return global_registry.get_plugin_class(plugin_name)

    def list_plugins(self) -> List[str]:
        """
        List all available plugins.
        """
        global_registry.load_all_plugins()
        new_plugins = global_registry.list_plugins()
        legacy_plugins = list(self._legacy_plugins.keys())

        # Remove duplicates, preferring new system
        all_plugins = list(dict.fromkeys(new_plugins + legacy_plugins))
        return all_plugins

__init__(logger=None)

Initialize the PluginManager.

Parameters:

Name Type Description Default
logger optional

Logger instance for debug/info/error messages.

 None
Source code in src/automax/core/managers/plugin_manager.py 
23
24
25
26
27
28
29
30
31
32
33
def __init__(self, logger=None):
    """
    Initialize the PluginManager.

    Args:
        logger (optional): Logger instance for debug/info/error messages.

    """
    self.logger = logger
    self._legacy_plugins: Dict[str, Any] = {}
    self._load_legacy_plugins()

execute_plugin(name, config)

Execute a plugin with given configuration.

This method first tries the new registry system, then falls back to legacy plugins for backward compatibility.

Source code in src/automax/core/managers/plugin_manager.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def execute_plugin(self, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Execute a plugin with given configuration.

    This method first tries the new registry system, then falls back to legacy
    plugins for backward compatibility.

    """
    if self.logger:
        self.logger.debug(f"Executing plugin {name} with configuration {config}")

    # Try new registry system first
    global_registry.load_all_plugins()

    if name in global_registry.list_plugins():
        try:
            plugin_class = global_registry.get_plugin_class(name)
            plugin_instance = plugin_class(config)
            return plugin_instance.execute()
        except Exception as e:
            raise PluginExecutionError(
                f"Plugin {name} execution failed: {e}"
            ) from e

    # Fall back to legacy system
    elif name in self._legacy_plugins:
        self.logger.warning(f"Using legacy plugin: {name}")
        try:
            return self._legacy_plugins[name](config)
        except Exception as e:
            raise PluginExecutionError(
                f"Legacy plugin {name} execution failed: {e}"
            ) from e

    else:
        available_plugins = global_registry.list_plugins() + list(
            self._legacy_plugins.keys()
        )
        raise PluginError(
            f"Plugin '{name}' not found. Available plugins: {available_plugins}"
        )

get_plugin(plugin_name)

Get plugin class by name.

Source code in src/automax/core/managers/plugin_manager.py 
112
113
114
115
116
def get_plugin(self, plugin_name: str):
    """
    Get plugin class by name.
    """
    return global_registry.get_plugin_class(plugin_name)

list_plugins()

List all available plugins.

Source code in src/automax/core/managers/plugin_manager.py 
118
119
120
121
122
123
124
125
126
127
128
def list_plugins(self) -> List[str]:
    """
    List all available plugins.
    """
    global_registry.load_all_plugins()
    new_plugins = global_registry.list_plugins()
    legacy_plugins = list(self._legacy_plugins.keys())

    # Remove duplicates, preferring new system
    all_plugins = list(dict.fromkeys(new_plugins + legacy_plugins))
    return all_plugins

get_plugin_manager()

Get the global plugin manager instance.

Source code in src/automax/core/managers/plugin_manager.py 
135
136
137
138
139
140
141
142
def get_plugin_manager() -> PluginManager:
    """
    Get the global plugin manager instance.
    """
    global _plugin_manager_instance
    if _plugin_manager_instance is None:
        _plugin_manager_instance = PluginManager()
    return _plugin_manager_instance

automax.core.managers.logger_manager

Logger manager for Automax.

Handles centralized logging to console, file, optional JSON, and error file.

AlignedFormatter

Bases: Formatter

Formatter that aligns level names and formats timestamps in ISO 8601 with local timezone.

Source code in src/automax/core/managers/logger_manager.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class AlignedFormatter(logging.Formatter):
    """
    Formatter that aligns level names and formats timestamps in ISO 8601 with local
    timezone.
    """

    def formatTime(self, record, datefmt=None):
        # Convert timestamp to ISO 8601 with local timezone and milliseconds
        dt = datetime.fromtimestamp(record.created).astimezone()
        return dt.isoformat(timespec="milliseconds")

    def format(self, record):
        # Align level name for console/file logs
        record.levelname = LEVEL_MAP.get(record.levelname, record.levelname).ljust(5)
        return super().format(record)

JsonStreamingFormatter

Bases: Formatter

Formatter that outputs log records as JSON objects compliant with RFC 8259.

Source code in src/automax/core/managers/logger_manager.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class JsonStreamingFormatter(logging.Formatter):
    """
    Formatter that outputs log records as JSON objects compliant with RFC 8259.
    """

    def format(self, record):
        log_entry = {
            "timestamp": datetime.fromtimestamp(record.created)
            .astimezone()
            .isoformat(timespec="milliseconds"),
            "level": logging.getLevelName(record.levelno).upper(),
            "message": record.getMessage(),
        }
        # Always add trailing comma
        return json.dumps(log_entry, ensure_ascii=False) + ","

LoggerManager

Manager class for centralized logging.

Source code in src/automax/core/managers/logger_manager.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
class LoggerManager:
    """
    Manager class for centralized logging.
    """

    def __init__(self, log_directory=None, log_level="DEBUG", json_log=False):
        """
        Initialize LoggerManager.

        Args:
            log_directory (str or Path, optional): Directory for log files.
            log_level (str): Logging level.
            json_log (bool): Enable JSON logging.

        """
        self.log_directory = Path(log_directory) if log_directory else Path("logs")
        self.log_directory.mkdir(parents=True, exist_ok=True)
        self.log_level = getattr(logging, log_level.upper(), logging.DEBUG)
        self.json_log = json_log
        self._logger_initialized = False

        self._setup_logger()

    @property
    def logger(self):
        return self._logger

    def _setup_logger(self):
        if self._logger_initialized:
            return

        # Generate timestamped filenames
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.log_file = self.log_directory / f"{LOGGER_NAME}_{timestamp}.log"
        self.error_file = self.log_directory / f"{LOGGER_NAME}_{timestamp}.err"
        self.json_file = (
            self.log_directory / f"{LOGGER_NAME}_{timestamp}.json"
            if self.json_log
            else None
        )

        # Initialize logger
        self._logger = logging.getLogger(LOGGER_NAME)
        self._logger.setLevel(self.log_level)

        # Console handler
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(
            AlignedFormatter("%(asctime)s [ %(levelname)s ]  %(message)s")
        )
        self._logger.addHandler(console_handler)

        # File handler
        file_handler = logging.FileHandler(self.log_file)
        file_handler.setFormatter(
            AlignedFormatter("%(asctime)s [ %(levelname)s ]  %(message)s")
        )
        self._logger.addHandler(file_handler)

        # JSON handler (if enabled)
        self.json_handler = None
        if self.json_log:
            self.json_handler = logging.FileHandler(self.json_file, mode="w")
            self.json_handler.setFormatter(JsonStreamingFormatter())
            self._logger.addHandler(self.json_handler)

        # Error handler (lazy: added on first ERROR/FATAL)
        self.error_handler = None

        # Shutdown at exit
        atexit.register(self.shutdown)
        self._logger_initialized = True

    def get_logger(self):
        """
        Return the internal logger instance.
        """
        return self.logger

    def _ensure_error_file_handler(self):
        """
        Ensure the error file handler is added if not already.
        """
        if not self.error_handler:
            self.error_handler = logging.FileHandler(self.error_file)
            self.error_handler.setLevel(logging.ERROR)
            self.error_handler.setFormatter(
                AlignedFormatter("%(asctime)s [ %(levelname)s ]  %(message)s")
            )
            self._logger.addHandler(self.error_handler)

    def _write_json_record(self, record):
        """
        Ensure the main log file exists (touch if needed).
        """
        if self.json_handler:
            self.json_handler.handle(record)

    def _ensure_log_file_exists(self):
        """
        Ensure the main log file exists (touch if needed).
        """
        self.log_file.touch(exist_ok=True)

    def _close_json_file(self):
        """
        Close JSON file handler if open.
        """
        if self.json_handler and not self.json_handler.stream.closed:
            self.json_handler.close()

    def shutdown(self):
        """Shutdown logger: flush and close all handlers."""
        for handler in self._logger.handlers[:]:
            try:
                if hasattr(handler.stream, "closed") and not handler.stream.closed:
                    handler.flush()
                    handler.close()
            except (AttributeError, ValueError):
                pass  # Skip if no stream or already closed
            self._logger.removeHandler(handler)

    # ----------------------
    # Convenience methods
    # ----------------------

    def debug(self, msg):
        """
        Log a message at DEBUG level.
        """
        self._logger.debug(msg)
        self._write_json_record(
            self._logger.makeRecord(
                LOGGER_NAME, logging.DEBUG, "", 0, msg, args=None, exc_info=None
            )
        )
        self._ensure_log_file_exists()

    def info(self, msg):
        """
        Log a message at INFO level.
        """
        self._logger.info(msg)
        self._write_json_record(
            self._logger.makeRecord(
                LOGGER_NAME, logging.INFO, "", 0, msg, args=None, exc_info=None
            )
        )
        self._ensure_log_file_exists()

    def warning(self, msg):
        """
        Log a message at WARNING level.
        """
        self._logger.warning(msg)
        self._write_json_record(
            self._logger.makeRecord(
                LOGGER_NAME, logging.WARNING, "", 0, msg, args=None, exc_info=None
            )
        )
        self._ensure_log_file_exists()

    def error(self, msg):
        """
        Log a message at ERROR level and write to .err file.
        """
        self._ensure_error_file_handler()
        self._logger.error(msg)
        self._write_json_record(
            self._logger.makeRecord(
                LOGGER_NAME, logging.ERROR, "", 0, msg, args=None, exc_info=None
            )
        )
        self._ensure_log_file_exists()

    def fatal(self, msg):
        """
        Log a message at FATAL level and write to .err file.
        """
        self._ensure_error_file_handler()
        self._logger.critical(msg)
        self._write_json_record(
            self._logger.makeRecord(
                LOGGER_NAME, logging.CRITICAL, "", 0, msg, args=None, exc_info=None
            )
        )
        self._ensure_log_file_exists()

    def log(self, level, msg):
        """
        Log a message at the given severity level.

        Args:
            level (str): Logging level ('DEBUG', 'INFO', 'WARN', 'ERROR', 'FATAL').
            msg (str): Message to log.

        """
        level = level.upper()
        if level == "FATAL":
            self.fatal(msg)
        elif level == "ERROR":
            self.error(msg)
        elif hasattr(self._logger, level.lower()):
            getattr(self._logger, level.lower())(msg)
            self._write_json_record(
                self._logger.makeRecord(
                    LOGGER_NAME,
                    getattr(logging, level),
                    "",
                    0,
                    msg,
                    args=None,
                    exc_info=None,
                )
            )
        else:
            self.info(msg)

__init__(log_directory=None, log_level='DEBUG', json_log=False)

Initialize LoggerManager.

Parameters:

Name Type Description Default
log_directory str or Path

Directory for log files.

 None
log_level str

Logging level.

 'DEBUG'
json_log bool

Enable JSON logging.

 False
Source code in src/automax/core/managers/logger_manager.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def __init__(self, log_directory=None, log_level="DEBUG", json_log=False):
    """
    Initialize LoggerManager.

    Args:
        log_directory (str or Path, optional): Directory for log files.
        log_level (str): Logging level.
        json_log (bool): Enable JSON logging.

    """
    self.log_directory = Path(log_directory) if log_directory else Path("logs")
    self.log_directory.mkdir(parents=True, exist_ok=True)
    self.log_level = getattr(logging, log_level.upper(), logging.DEBUG)
    self.json_log = json_log
    self._logger_initialized = False

    self._setup_logger()

debug(msg)

Log a message at DEBUG level.

Source code in src/automax/core/managers/logger_manager.py 
190
191
192
193
194
195
196
197
198
199
200
def debug(self, msg):
    """
    Log a message at DEBUG level.
    """
    self._logger.debug(msg)
    self._write_json_record(
        self._logger.makeRecord(
            LOGGER_NAME, logging.DEBUG, "", 0, msg, args=None, exc_info=None
        )
    )
    self._ensure_log_file_exists()

error(msg)

Log a message at ERROR level and write to .err file.

Source code in src/automax/core/managers/logger_manager.py 
226
227
228
229
230
231
232
233
234
235
236
237
def error(self, msg):
    """
    Log a message at ERROR level and write to .err file.
    """
    self._ensure_error_file_handler()
    self._logger.error(msg)
    self._write_json_record(
        self._logger.makeRecord(
            LOGGER_NAME, logging.ERROR, "", 0, msg, args=None, exc_info=None
        )
    )
    self._ensure_log_file_exists()

fatal(msg)

Log a message at FATAL level and write to .err file.

Source code in src/automax/core/managers/logger_manager.py 
239
240
241
242
243
244
245
246
247
248
249
250
def fatal(self, msg):
    """
    Log a message at FATAL level and write to .err file.
    """
    self._ensure_error_file_handler()
    self._logger.critical(msg)
    self._write_json_record(
        self._logger.makeRecord(
            LOGGER_NAME, logging.CRITICAL, "", 0, msg, args=None, exc_info=None
        )
    )
    self._ensure_log_file_exists()

get_logger()

Return the internal logger instance.

Source code in src/automax/core/managers/logger_manager.py 
137
138
139
140
141
def get_logger(self):
    """
    Return the internal logger instance.
    """
    return self.logger

info(msg)

Log a message at INFO level.

Source code in src/automax/core/managers/logger_manager.py 
202
203
204
205
206
207
208
209
210
211
212
def info(self, msg):
    """
    Log a message at INFO level.
    """
    self._logger.info(msg)
    self._write_json_record(
        self._logger.makeRecord(
            LOGGER_NAME, logging.INFO, "", 0, msg, args=None, exc_info=None
        )
    )
    self._ensure_log_file_exists()

log(level, msg)

Log a message at the given severity level.

Parameters:

Name Type Description Default
level str

Logging level ('DEBUG', 'INFO', 'WARN', 'ERROR', 'FATAL').

 required
msg str

Message to log.

 required
Source code in src/automax/core/managers/logger_manager.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def log(self, level, msg):
    """
    Log a message at the given severity level.

    Args:
        level (str): Logging level ('DEBUG', 'INFO', 'WARN', 'ERROR', 'FATAL').
        msg (str): Message to log.

    """
    level = level.upper()
    if level == "FATAL":
        self.fatal(msg)
    elif level == "ERROR":
        self.error(msg)
    elif hasattr(self._logger, level.lower()):
        getattr(self._logger, level.lower())(msg)
        self._write_json_record(
            self._logger.makeRecord(
                LOGGER_NAME,
                getattr(logging, level),
                "",
                0,
                msg,
                args=None,
                exc_info=None,
            )
        )
    else:
        self.info(msg)

shutdown()

Shutdown logger: flush and close all handlers.

Source code in src/automax/core/managers/logger_manager.py 
175
176
177
178
179
180
181
182
183
184
def shutdown(self):
    """Shutdown logger: flush and close all handlers."""
    for handler in self._logger.handlers[:]:
        try:
            if hasattr(handler.stream, "closed") and not handler.stream.closed:
                handler.flush()
                handler.close()
        except (AttributeError, ValueError):
            pass  # Skip if no stream or already closed
        self._logger.removeHandler(handler)

warning(msg)

Log a message at WARNING level.

Source code in src/automax/core/managers/logger_manager.py 
214
215
216
217
218
219
220
221
222
223
224
def warning(self, msg):
    """
    Log a message at WARNING level.
    """
    self._logger.warning(msg)
    self._write_json_record(
        self._logger.makeRecord(
            LOGGER_NAME, logging.WARNING, "", 0, msg, args=None, exc_info=None
        )
    )
    self._ensure_log_file_exists()