Plugins

Base classes, registry, errors, and template helpers for the Imbi plugin system. Plugin authors should pair this reference with the Authoring Plugins guide, which explains the plugin variations and the contract each must satisfy.

Manifest and Context

PluginManifest

Bases: BaseModel

PluginOption

Bases: BaseModel

CredentialField

Bases: BaseModel

DataType

Bases: BaseModel

PluginContext

Bases: BaseModel

Configuration Plugins

The configuration variation models a typed key/value store scoped to a project. Implementations subclass ConfigurationPlugin and declare plugin_type='configuration' in their manifest.

ConfigurationPlugin

Bases: ABC

Plugins must not stash global state.

A new instance is created per request.

ConfigKey

Bases: BaseModel

ConfigKeyWithValue

Bases: ConfigKey

ConfigValue

Bases: BaseModel

Logs Plugins

The logs variation exposes a search interface against an external log store. Implementations subclass LogsPlugin and declare plugin_type='logs' in their manifest.

LogsPlugin

Bases: ABC

histogram async

histogram(
    ctx: PluginContext,
    credentials: dict[str, str],
    query: LogQuery,
    bucket_count: int = 60,
) -> list[LogHistogramBucket]

Return time-bucketed event counts for the histogram view.

Plugins that support histograms should override this method and set manifest.supports_histogram = True. The default implementation returns an empty list, which causes the API to signal that histograms are unavailable for this source.

Source code in src/imbi_common/plugins/base.py

async def histogram(
    self,
    ctx: PluginContext,
    credentials: dict[str, str],
    query: LogQuery,
    bucket_count: int = 60,
) -> list[LogHistogramBucket]:
    """Return time-bucketed event counts for the histogram view.

    Plugins that support histograms should override this method and
    set ``manifest.supports_histogram = True``.  The default
    implementation returns an empty list, which causes the API to
    signal that histograms are unavailable for this source.
    """
    return []

LogQuery

Bases: BaseModel

LogFilter

Bases: BaseModel

LogEntry

Bases: BaseModel

LogResult

Bases: BaseModel

LogHistogramBucket

Bases: BaseModel

A single time bucket in a log histogram response.

Webhook Action Plugins

The webhook variation reacts to inbound webhook payloads routed by a host such as imbi-gateway. Implementations subclass WebhookActionPlugin and declare plugin_type='webhook' in their manifest. The host parses the WebhookRule.handler field as "<plugin_slug>:<action_name>", fetches per-instance credentials via get_plugin_credentials (see below), and dispatches by calling run_action.

WebhookActionPlugin

Bases: ABC

Base class for webhook-action plugins.

Webhook action plugins declare a manifest (slug, credentials, optional configuration options) so operators can install and configure them like any other plugin, and advertise a static catalog of actions through :meth:actions. The host (imbi-gateway) parses WebhookRule.handler as "<plugin_slug>#<action_name>", looks the plugin up in the registry, picks the matching :class:ActionDescriptor, validates the rule's handler_config against :attr:ActionDescriptor.config_model, and calls :attr:ActionDescriptor.callable with the uniform signature captured in :data:WebhookActionCallable.

The plugin class itself carries no runtime dispatch logic -- the callable lives wherever the descriptor points (typically the plugin's own actions submodule).

actions abstractmethod classmethod

actions() -> list[ActionDescriptor]

Return the static catalog of actions this plugin exposes.

Implementations should return a fresh list each call so callers can mutate the result safely. The host validates each descriptor's callable and config_model ImportStrings at construction time, so misconfigured paths fail loud during registry load rather than at request time.

Source code in src/imbi_common/plugins/base.py

@classmethod
@abc.abstractmethod
def actions(cls) -> list[ActionDescriptor]:
    """Return the static catalog of actions this plugin exposes.

    Implementations should return a fresh list each call so callers
    can mutate the result safely. The host validates each
    descriptor's ``callable`` and ``config_model`` ImportStrings at
    construction time, so misconfigured paths fail loud during
    registry load rather than at request time.
    """

Credential Resolution

Helpers for hosts that need to fetch decrypted plugin credentials out of the graph. Routing depends on manifest.auth_type: api_token and aws-iam-ic read the Fernet-encrypted Plugin.plugin_configuration blob directly, while oauth2 and oidc walk Plugin -[:USES_APPLICATION]-> ServiceApplication.

get_plugin_credentials async

get_plugin_credentials(
    db: Graph, plugin_id: str, entry: RegistryEntry
) -> dict[str, str]

Fetch and decrypt plugin credentials from the graph.

Raises:

Type	Description
PluginCredentialsMissing	If required credentials are absent or the plugin is not linked to a ServiceApplication.

Source code in src/imbi_common/plugins/credentials.py

async def get_plugin_credentials(
    db: graph.Graph,
    plugin_id: str,
    entry: RegistryEntry,
) -> dict[str, str]:
    """Fetch and decrypt plugin credentials from the graph.

    Raises:
        PluginCredentialsMissing: If required credentials are absent or
            the plugin is not linked to a ServiceApplication.
    """
    auth_type = entry.manifest.auth_type
    if auth_type in ('api_token', 'aws-iam-ic'):
        return await _get_plugin_configuration_credentials(
            db, plugin_id, entry
        )
    return await _get_application_credentials(db, plugin_id, entry)

get_plugin_configuration_keys async

get_plugin_configuration_keys(
    db: Graph, plugin_id: str
) -> list[str]

Return the set of credential field names currently set.

The plaintext values themselves are never surfaced to the caller.

Source code in src/imbi_common/plugins/credentials.py

async def get_plugin_configuration_keys(
    db: graph.Graph,
    plugin_id: str,
) -> list[str]:
    """Return the set of credential field names currently set.

    The plaintext values themselves are never surfaced to the caller.
    """
    try:
        data = await _read_plugin_configuration(db, plugin_id)
    except ValueError:
        LOGGER.warning(
            'plugin_configuration unreadable for plugin_id=%s; '
            'reporting no keys',
            plugin_id,
        )
        return []
    return [k for k, v in data.items() if v]

patch_plugin_configuration async

patch_plugin_configuration(
    db: Graph,
    plugin_id: str,
    updates: dict[str, str | None],
) -> list[str]

Apply a partial update to plugin_configuration.

updates maps field name to new plaintext value. None (or empty string) removes the field. Existing keys not present in updates are preserved. Returns the resulting set of populated keys (no plaintext values).

Source code in src/imbi_common/plugins/credentials.py

async def patch_plugin_configuration(
    db: graph.Graph,
    plugin_id: str,
    updates: dict[str, str | None],
) -> list[str]:
    """Apply a partial update to ``plugin_configuration``.

    ``updates`` maps field name to new plaintext value. ``None`` (or
    empty string) removes the field. Existing keys not present in
    ``updates`` are preserved. Returns the resulting set of populated
    keys (no plaintext values).
    """
    current = await _read_plugin_configuration(db, plugin_id)
    for key, value in updates.items():
        if value is None or value == '':
            current.pop(key, None)
        else:
            current[key] = value
    encrypted = TokenEncryption.get_instance().encrypt(json.dumps(current))
    query: typing.LiteralString = """
    MATCH (p:Plugin {{id: {plugin_id}}})
    SET p.plugin_configuration = {blob}
    RETURN p
    """
    await db.execute(
        query,
        {'plugin_id': plugin_id, 'blob': encrypted},
        [],
    )
    return [k for k, v in current.items() if v]

Registry

Plugin discovery is driven by importlib.metadata entry points under the imbi.plugins group. The host calls load_plugins() at startup (and reload_plugins() on demand) to populate the registry.

load_plugins

load_plugins() -> LoadResult

Discover and load all installed imbi plugins.

Source code in src/imbi_common/plugins/registry.py

def load_plugins() -> LoadResult:
    """Discover and load all installed imbi plugins."""
    loaded: list[str] = []
    errors: dict[str, str] = {}
    skipped: list[str] = []
    new_registry: dict[str, RegistryEntry] = {}

    for ep in importlib.metadata.entry_points(group=_ENTRY_POINT_GROUP):
        try:
            cls = ep.load()
        except Exception as exc:
            LOGGER.exception('Failed to load plugin %r: %s', ep.name, exc)
            errors[ep.name] = str(exc)
            continue

        if not hasattr(cls, 'manifest') or not isinstance(
            cls.manifest, PluginManifest
        ):
            LOGGER.error(
                'Plugin %r has no valid manifest attribute; skipping',
                ep.name,
            )
            errors[ep.name] = 'Missing or invalid manifest attribute'
            continue

        manifest: PluginManifest = cls.manifest

        if not isinstance(cls, type) or not issubclass(
            cls,
            (
                ConfigurationPlugin,
                LogsPlugin,
                IdentityPlugin,
                DeploymentPlugin,
                LifecyclePlugin,
                WebhookActionPlugin,
            ),
        ):
            LOGGER.error(
                'Plugin %r does not implement ConfigurationPlugin, '
                'LogsPlugin, IdentityPlugin, DeploymentPlugin, '
                'LifecyclePlugin, or WebhookActionPlugin; skipping',
                ep.name,
            )
            errors[ep.name] = (
                'Plugin must subclass ConfigurationPlugin, LogsPlugin, '
                'IdentityPlugin, DeploymentPlugin, LifecyclePlugin, or '
                'WebhookActionPlugin'
            )
            continue

        expected_base = _PLUGIN_TYPE_BASES[manifest.plugin_type]
        if not issubclass(cls, expected_base):  # pyright: ignore[reportUnnecessaryIsInstance]
            LOGGER.error(
                'Plugin %r manifest plugin_type=%r does not match class '
                'hierarchy; skipping',
                ep.name,
                manifest.plugin_type,
            )
            errors[ep.name] = (
                f'Class does not implement {expected_base.__name__} '
                f'for plugin_type={manifest.plugin_type!r}'
            )
            continue

        if manifest.api_version not in _SUPPORTED_API_VERSIONS:
            LOGGER.warning(
                'Plugin %r declares api_version=%d not in supported '
                'versions %s; skipping',
                ep.name,
                manifest.api_version,
                sorted(_SUPPORTED_API_VERSIONS),
            )
            skipped.append(ep.name)
            continue

        if issubclass(cls, WebhookActionPlugin):
            try:
                descriptors = cls.actions()
            except Exception as exc:
                LOGGER.exception(
                    'Plugin %r failed to enumerate actions: %s',
                    ep.name,
                    exc,
                )
                errors[ep.name] = f'actions() raised: {exc}'
                continue
            if not isinstance(descriptors, list) or any(
                not isinstance(descriptor, ActionDescriptor)
                for descriptor in descriptors
            ):
                LOGGER.error(
                    'Plugin %r actions() did not return '
                    'list[ActionDescriptor]; skipping',
                    ep.name,
                )
                errors[ep.name] = (
                    'actions() must return list[ActionDescriptor]'
                )
                continue
            action_error = _validate_action_catalog(ep.name, descriptors)
            if action_error is not None:
                errors[ep.name] = action_error
                continue

        dist = ep.dist
        pkg_name = dist.name if dist else 'unknown'
        pkg_version = dist.version if dist else 'unknown'

        entry = RegistryEntry(
            handler_cls=cls,
            manifest=manifest,
            package_name=pkg_name,
            package_version=pkg_version,
        )
        if manifest.slug in new_registry:
            LOGGER.error(
                'Duplicate plugin slug %r from entry point %r; skipping',
                manifest.slug,
                ep.name,
            )
            errors[ep.name] = f'Duplicate plugin slug: {manifest.slug}'
            continue
        new_registry[manifest.slug] = entry
        LOGGER.info(
            'Loaded plugin %r v%s (slug=%r, api_version=%d)',
            pkg_name,
            pkg_version,
            manifest.slug,
            manifest.api_version,
        )
        loaded.append(manifest.slug)

    # Refuse vlabel/edge collisions across loaded plugins or with core
    # schemata.  Imported lazily to avoid a circular import.
    from imbi_common.plugins.schemas import validate_no_collisions

    validate_no_collisions([entry.manifest for entry in new_registry.values()])

    with _lock:
        _registry.clear()
        _registry.update(new_registry)

    return LoadResult(loaded=loaded, errors=errors, skipped=skipped)

reload_plugins

reload_plugins() -> LoadResult

Reload the plugin registry from installed entry points.

Source code in src/imbi_common/plugins/registry.py

def reload_plugins() -> LoadResult:
    """Reload the plugin registry from installed entry points."""
    LOGGER.info('Reloading plugin registry')
    return load_plugins()

get_plugin

get_plugin(slug: str) -> RegistryEntry

Get a registry entry by plugin slug.

Raises:

Type	Description
PluginNotFoundError	If the slug is not registered.

Source code in src/imbi_common/plugins/registry.py

def get_plugin(slug: str) -> RegistryEntry:
    """Get a registry entry by plugin slug.

    Raises:
        PluginNotFoundError: If the slug is not registered.
    """
    with _lock:
        entry = _registry.get(slug)
    if entry is None:
        raise PluginNotFoundError(slug)
    return entry

list_plugins

list_plugins() -> list[RegistryEntry]

Return all registered plugins.

Source code in src/imbi_common/plugins/registry.py

def list_plugins() -> list[RegistryEntry]:
    """Return all registered plugins."""
    with _lock:
        return list(_registry.values())

RegistryEntry dataclass

RegistryEntry(
    handler_cls: type[ConfigurationPlugin]
    | type[LogsPlugin]
    | type[IdentityPlugin]
    | type[DeploymentPlugin]
    | type[LifecyclePlugin]
    | type[WebhookActionPlugin],
    manifest: PluginManifest,
    package_name: str,
    package_version: str,
)

LoadResult

Bases: NamedTuple

Errors

PluginNotFoundError

Bases: Exception

Raised when a plugin slug is not registered.

PluginUnavailableError

Bases: Exception

Raised when a plugin slug exists in the graph but not the registry.

PluginCredentialsMissing

Bases: Exception

Raised when required credentials are absent for a plugin.

PluginTimeoutError

Bases: Exception

Raised when a plugin call exceeds the configured timeout.

CursorExpiredError

Bases: Exception

Raised by log plugins when a pagination cursor has expired.

Templates

Helpers for plugins that build provider-specific query strings from project context. Substitution is restricted to a fixed whitelist of variables (project_slug, org_slug, environment, project_id); unknown variables raise ValueError.

validate_template

validate_template(template: str) -> None

Validate a search template string, rejecting unknown variables.

Raises:

Type	Description
ValueError	If the template references variables not in the whitelist.

Source code in src/imbi_common/plugins/templates.py

def validate_template(template: str) -> None:
    """Validate a search template string, rejecting unknown variables.

    Raises:
        ValueError: If the template references variables not in the whitelist.
    """
    for match in _VAR_PATTERN.finditer(template):
        var = match.group(1)
        if var not in _ALLOWED_VARS:
            raise ValueError(
                f'Unknown template variable ${{{var}}};'
                f' allowed: {sorted(_ALLOWED_VARS)}'
            )

expand_template

expand_template(
    template: str, variables: dict[str, str | None]
) -> str

Expand a search template substituting whitelisted variables.

Absent variable values are replaced with empty strings. The template is validated first to reject unknown placeholders, preserving the whitelist guarantee from :func:validate_template.

Source code in src/imbi_common/plugins/templates.py

def expand_template(
    template: str,
    variables: dict[str, str | None],
) -> str:
    """Expand a search template substituting whitelisted variables.

    Absent variable values are replaced with empty strings. The template is
    validated first to reject unknown placeholders, preserving the whitelist
    guarantee from :func:`validate_template`.
    """
    validate_template(template)

    def _sub(match: re.Match[str]) -> str:
        var = match.group(1)
        val = variables.get(var)
        return val if val is not None else ''

    return _VAR_PATTERN.sub(_sub, template)