Plugin System Reference

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')

class apilinker.core.plugins.AuthPlugin(**kwargs: Any)[source]

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][source]

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

plugin_type: str = 'auth'

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

Validate that the credentials dictionary has the required fields.

Parameters:: credentials – Credentials dictionary to validate
Returns:: True if valid, False otherwise

class apilinker.core.plugins.ConnectorPlugin(**kwargs: Any)[source]

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()

connect(**kwargs) → Any[source]

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[source]

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[source]

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[source]

Validate that a connection object is properly formed.

Parameters:: connection – Connection object to validate
Returns:: True if valid, False otherwise

class apilinker.core.plugins.PluginBase(**kwargs: Any)[source]

Bases: object

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[source]

Initialize the plugin with the provided configuration.

Parameters:: **kwargs – Configuration parameters for the plugin

classmethod get_plugin_info() → Dict[str, Any][source]

Get metadata about this plugin.

Returns:: Dictionary containing plugin metadata (type, name, description)

plugin_name: str = 'base'

plugin_type: str = 'base'

exception apilinker.core.plugins.PluginError[source]

Bases: Exception

Base exception for all plugin-related errors.

exception apilinker.core.plugins.PluginInitializationError[source]

Bases: PluginError

Raised when a plugin cannot be initialized.

class apilinker.core.plugins.PluginManager[source]

Bases: object

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__()[source]: Initialize the plugin manager with empty plugin registries.

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

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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

exception apilinker.core.plugins.PluginNotFoundError[source]

Bases: PluginError

Raised when a requested plugin cannot be found.

exception apilinker.core.plugins.PluginValidationError[source]

Bases: PluginError

Raised when a plugin fails validation checks.

class apilinker.core.plugins.TransformerPlugin(**kwargs: Any)[source]

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

plugin_type: str = 'transformer'

transform(value: Any, **kwargs) → Any[source]

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[source]

Validate that the input value is appropriate for this transformer.

Parameters:: value – Input value to validate
Returns:: True if valid, False otherwise

Overview

ApiLinker’s plugin system allows you to extend functionality through custom plugins for data transformation, API connections, and authentication methods. The system is designed to be flexible, type-safe, and easy to use.

Plugin Types

ApiLinker supports three main types of plugins:

Transformer Plugins: Convert data between formats, validate values, or perform calculations
Connector Plugins: Handle communication with different API types
Auth Plugins: Implement various authentication methods

Creating Custom Plugins

To create a custom plugin, inherit from one of the base classes and implement the required methods:

from apilinker.core.plugins import TransformerPlugin

class PhoneNumberFormatter(TransformerPlugin):
    """Format phone numbers to E.164 international format."""

    plugin_name = "phone_formatter"

    def transform(self, value, **kwargs):
        if not value:
            return None

        # Remove non-digits
        digits = ''.join(c for c in value if c.isdigit())

        # Format based on length
        if len(digits) == 10:  # US number without country code
            return f"+1{digits}"
        elif len(digits) > 10:  # Assume international number
            return f"+{digits}"
        else:
            return value  # Return original if can't format

Plugin Discovery

ApiLinker automatically discovers plugins from several locations:

Built-in plugins in apilinker.plugins.builtin
User plugins in ~/.apilinker/plugins/
Package plugins in apilinker/plugins/
Custom directories specified by the user

Using Plugins

Once a plugin is registered, it can be used in configuration files or programmatically:

# In YAML config
mapping:
  - source: user.phone
    target: contact.phoneNumber
    transform: phone_formatter

# In Python code
from apilinker import ApiLinker

linker = ApiLinker()

# Use transformer directly
transformer = linker.plugin_manager.get_transformer("phone_formatter")
formatted = transformer("+1 (555) 123-4567")

# Use auth plugin
auth = linker.plugin_manager.get_auth_plugin("oauth2")
credentials = auth.authenticate(
    client_id="client_id",
    client_secret="client_secret",
    token_url="https://auth.example.com/token"
)

Error Handling

The plugin system includes comprehensive error handling:

PluginNotFoundError: Raised when a requested plugin cannot be found
PluginValidationError: Raised when a plugin fails validation checks
PluginInitializationError: Raised when a plugin cannot be initialized

Plugins should also include their own validation and error handling to ensure robust operation and clear error messages.