Plugins Module

Plugin system for extending ApiLinker functionality.

This module provides the foundation for ApiLinker’s extensibility through a plugin architecture. It includes base classes for different plugin types (transformers, connectors, and authentication) and a plugin manager for discovery and loading of plugins.

Example

>>> from apilinker.core.plugins import PluginManager
>>> manager = PluginManager()
>>> manager.discover_plugins()
>>> transformer = manager.get_transformer('my_transformer')
>>> result = transformer('input value')

Plugin Base Classes

PluginBase

class apilinker.core.plugins.PluginBase(**kwargs: Any)

Base class for all ApiLinker plugins.

All plugins must inherit from this class or one of its subclasses and provide required implementation details.

plugin_type

Type of the plugin (e.g., transformer, connector, auth)

Type:

str

plugin_name

Unique name of the plugin

Type:

str

config

Configuration parameters passed during initialization

Type:

dict

__init__(**kwargs: Any) → None

Initialize the plugin with the provided configuration.

Parameters:

**kwargs – Configuration parameters for the plugin

config: Dict[str, Any]

classmethod get_plugin_info() → Dict[str, Any]

Get metadata about this plugin.

Returns:

Dictionary containing plugin metadata (type, name, description)

plugin_name: str = 'base'

plugin_type: str = 'base'

TransformerPlugin

class apilinker.core.plugins.TransformerPlugin(**kwargs: Any)

Bases: PluginBase

Base class for data transformation plugins.

Transformer plugins are used to convert, format, or validate data during the mapping process between source and target APIs.

Example

class LowercaseTransformer(TransformerPlugin):
    plugin_name = "lowercase"

    def transform(self, value: str, **kwargs) -> str:
        return value.lower() if isinstance(value, str) else value

config: Dict[str, Any]

plugin_type: str = 'transformer'

transform(value: Any, **kwargs) → Any

Transform a value according to the plugin’s logic.

Parameters:

• value – Input value to transform

• **kwargs – Additional parameters for customizing the transformation

Returns:

Transformed value

Raises:

• NotImplementedError – If the subclass does not implement this method

• ValueError – If the input value is not valid for this transformer

• TypeError – If the input value is of an incompatible type

validate_input(value: Any) → bool

Validate that the input value is appropriate for this transformer.

Parameters:

value – Input value to validate

Returns:

True if valid, False otherwise

ConnectorPlugin

class apilinker.core.plugins.ConnectorPlugin(**kwargs: Any)

Bases: PluginBase

Base class for API connector plugins.

Connector plugins handle the communication with external APIs, including connection management, data fetching, and sending.

Example

class RestConnector(ConnectorPlugin):
    plugin_name = "rest"

    def connect(self, base_url: str, **kwargs) -> dict:
        session = requests.Session()
        return {"session": session, "base_url": base_url}

    def fetch(self, connection: dict, endpoint: str, **kwargs) -> dict:
        url = f"{connection['base_url']}/{endpoint}"
        response = connection['session'].get(url, **kwargs)
        response.raise_for_status()
        return response.json()

config: Dict[str, Any]

connect(**kwargs) → Any

Create a connection to the API.

This method should establish a connection or create a client that can be used for subsequent API operations.

Parameters:

**kwargs – Connection parameters like base_url, timeout, etc.

Returns:

Connection object that will be passed to fetch and send methods

Raises:

• NotImplementedError – If the subclass does not implement this method

• ConnectionError – If the connection cannot be established

• ValueError – If required parameters are missing or invalid

fetch(connection: Any, endpoint: str, **kwargs) → Any

Fetch data from the API.

This method should retrieve data from the specified API endpoint using the previously established connection.

Parameters:

• connection – Connection object from connect()

• endpoint – Endpoint path or identifier to fetch from

• **kwargs – Additional parameters like query params, headers, etc.

Returns:

Fetched data, typically parsed from JSON response

Raises:

• NotImplementedError – If the subclass does not implement this method

• ConnectionError – If the request fails due to network issues

• ValueError – If the endpoint or parameters are invalid

• Exception – Other API-specific errors that might occur

plugin_type: str = 'connector'

send(connection: Any, endpoint: str, data: Any, **kwargs) → Any

Send data to the API.

This method should send data to the specified API endpoint using the previously established connection.

Parameters:

• connection – Connection object from connect()

• endpoint – Endpoint path or identifier to send to

• data – Data to send (typically a dict that will be serialized to JSON)

• **kwargs – Additional parameters like headers, query params, etc.

Returns:

API response, typically parsed from JSON response

Raises:

• NotImplementedError – If the subclass does not implement this method

• ConnectionError – If the request fails due to network issues

• ValueError – If the endpoint, data, or parameters are invalid

• Exception – Other API-specific errors that might occur

validate_connection(connection: Any) → bool

Validate that a connection object is properly formed.

Parameters:

connection – Connection object to validate

Returns:

True if valid, False otherwise

AuthPlugin

class apilinker.core.plugins.AuthPlugin(**kwargs: Any)

Bases: PluginBase

Base class for authentication plugins.

Authentication plugins handle different methods of API authentication, such as API keys, OAuth2, JWT tokens, etc.

Example

class BearerTokenAuth(AuthPlugin):
    plugin_name = "bearer"

    def authenticate(self, token: str, **kwargs) -> Dict[str, Any]:
        if not token:
            raise ValueError("Bearer token cannot be empty")
        return {"type": "bearer", "headers": {"Authorization": f"Bearer {token}"}}

authenticate(**kwargs) → Dict[str, Any]

Perform authentication and return credentials.

This method should implement the authentication logic and return credentials in a format that can be used by connectors.

Parameters:

**kwargs – Authentication parameters (tokens, keys, etc.)

Returns:

• headers: Dict of HTTP headers to include in requests

• params: Dict of URL parameters to include (optional)

• auth: Auth object for requests library (optional)

• type: String identifier of the auth type

Return type:

Dictionary containing authentication result with at least

Raises:

• NotImplementedError – If the subclass does not implement this method

• ValueError – If required parameters are missing or invalid

• AuthenticationError – If authentication fails

config: Dict[str, Any]

plugin_type: str = 'auth'

validate_credentials(credentials: Dict[str, Any]) → bool

Validate that the credentials dictionary has the required fields.

Parameters:

credentials – Credentials dictionary to validate

Returns:

True if valid, False otherwise

Plugin Management

PluginManager

class apilinker.core.plugins.PluginManager

Manager for loading and using plugins.

This class handles the discovery, loading, and management of plugins for extending ApiLinker functionality.

Example

# Initialize the manager
manager = PluginManager()

# Discover available plugins
discovered = manager.discover_plugins()
print(f"Found {len(discovered)} plugins")

# Register a custom plugin class
manager.register_plugin(MyCustomTransformer)

# Get a transformer instance
transformer = manager.instantiate_plugin("transformer", "my_custom")
result = transformer.transform("input data")

__init__()

Initialize the plugin manager with empty plugin registries.

discover_plugins(plugin_dir: str | None = None) → List[Dict[str, Any]]

Discover available plugins from built-in and custom directories.

This method searches for Python modules containing plugin classes in the built-in plugins directory and optional custom directories. Discovered plugins are automatically registered with the manager.

Parameters:

plugin_dir – Optional custom directory to search for plugins. If None, default locations will be searched.

Returns:

List of dictionaries containing information about discovered plugins

Raises:

• PermissionError – If a plugin directory exists but cannot be read

• ImportError – If a plugin module exists but cannot be imported

get_auth_plugin(name: str, **kwargs) → AuthPlugin | None

Get an authentication plugin instance.

Parameters:

• name – Name of the authentication plugin

• **kwargs – Plugin initialization parameters

Returns:

Auth plugin instance or None if not found

Raises:

• PluginNotFoundError – If the auth plugin cannot be found

• PluginInitializationError – If the plugin cannot be initialized

• TypeError – If the plugin is not an AuthPlugin

get_connector(name: str, **kwargs) → ConnectorPlugin | None

Get a connector plugin instance.

Parameters:

• name – Name of the connector plugin

• **kwargs – Plugin initialization parameters

Returns:

Connector plugin instance or None if not found

Raises:

• PluginNotFoundError – If the connector plugin cannot be found

• PluginInitializationError – If the plugin cannot be initialized

• TypeError – If the plugin is not a ConnectorPlugin

get_plugin(plugin_type: str, plugin_name: str) → Type[PluginBase] | None

Get a plugin class by type and name.

Parameters:

• plugin_type – Type of plugin (transformer, connector, auth, etc.)

• plugin_name – Name of the specific plugin

Returns:

Plugin class if found, None otherwise

Raises:

TypeError – If plugin_type or plugin_name are not strings

get_transformer(name: str, **kwargs) → Callable[[Any], Any] | None

Get a transformer function from a plugin.

This is a convenience method that returns a callable function wrapping the transform method of a TransformerPlugin instance.

Parameters:

• name – Name of the transformer plugin

• **kwargs – Plugin initialization parameters

Returns:

Callable function that accepts a value and parameters and returns transformed value

Raises:

• PluginNotFoundError – If the transformer plugin cannot be found

• PluginInitializationError – If the plugin cannot be initialized

• TypeError – If the plugin is not a TransformerPlugin

instantiate_plugin(plugin_type: str, plugin_name: str, **kwargs) → PluginBase | None

Create an instance of a plugin with the given parameters.

Parameters:

• plugin_type – Type of plugin to instantiate

• plugin_name – Name of plugin to instantiate

• **kwargs – Parameters to pass to the plugin constructor

Returns:

Instantiated plugin object if successful, None otherwise

Raises:

• PluginNotFoundError – If the plugin cannot be found

• PluginInitializationError – If the plugin cannot be initialized

register_plugin(plugin_class: Type[PluginBase]) → None

Register a plugin class with the manager.

Parameters:

plugin_class – Plugin class to register, must be a subclass of PluginBase

Raises:

• PluginValidationError – If the plugin class is invalid

• TypeError – If the provided class is not a PluginBase subclass

Exceptions

class apilinker.core.plugins.PluginError

Bases: Exception

Base exception for all plugin-related errors.

class apilinker.core.plugins.PluginNotFoundError

Bases: PluginError

Raised when a requested plugin cannot be found.

class apilinker.core.plugins.PluginValidationError

Bases: PluginError

Raised when a plugin fails validation checks.

class apilinker.core.plugins.PluginInitializationError

Bases: PluginError

Raised when a plugin cannot be initialized.