Snakemake Logger Plugin Interface

Common errors

• TypeError: Can't instantiate abstract class MyLoggerPlugin with abstract methods get_wrapper, register_args, setup

  cause Your custom logger plugin class `MyLoggerPlugin` (or `MyLoggerWrapper`) is missing implementations for one or more abstract methods inherited from `LoggerPlugin` or `LoggerWrapper`.
  fix Implement all abstract methods defined in `snakemake_interface_logger_plugins.plugin.LoggerPlugin` and `snakemake_interface_logger_plugins.wrapper.LoggerWrapper` in your custom classes.

• snakemake.exceptions.PluginError: Could not find logger plugin 'my_non_existent_plugin.py'

  cause The Python file or module specified with `--logger-plugin` could not be found or imported by Snakemake, or it does not contain a class named `LoggerPlugin`.
  fix Verify the `--logger-plugin` argument points to the correct file path or an importable Python module. Ensure the file exists and is accessible, or that the module is correctly installed and discoverable in the Python environment. Also, ensure your plugin class is named `LoggerPlugin` in the specified file/module.

Warnings

• breaking The logger plugin API underwent a significant redesign with Snakemake 8.0. Versions of `snakemake-interface-logger-plugins` prior to 2.0.0 (e.g., 1.x) are not compatible with Snakemake 8.0+.
  Fix: Upgrade `snakemake-interface-logger-plugins` to version 2.0.0 or higher to match Snakemake 8.0+ requirements. If targeting older Snakemake versions, pin `snakemake-interface-logger-plugins` to a compatible 1.x version.
• gotcha Snakemake's plugin discovery requires the `--logger-plugin` argument to point to a valid Python file (e.g., `my_plugin.py`) or a discoverable Python module (e.g., `my_package.my_module`). Incorrect paths or non-discoverable modules will prevent the plugin from loading.
  Fix: Ensure the path provided to `--logger-plugin` is correct and the Python file/module is accessible from where Snakemake is run. For packages, ensure the plugin module is importable and that it contains a class named `LoggerPlugin`.
• gotcha When implementing `LoggerPlugin` or `LoggerWrapper`, you must implement all abstract methods defined in the interface. Failing to do so will result in `TypeError` when attempting to instantiate your plugin.
  Fix: Refer to the `LoggerPlugin` and `LoggerWrapper` base classes for the full list of abstract methods (e.g., `handle_info`, `handle_error`, `handle_job_info`, `setup`, `register_args`, `get_wrapper`). Implement all of them, even if with a `pass` statement, if no specific action is needed.

Install

• pip install snakemake-interface-logger-plugins Install core interface
• pip install snakemake Install Snakemake (required for use)

Imports

• LoggerPlugin

  from snakemake_interface_logger_plugins.plugin import LoggerPlugin

• LoggerWrapper

  from snakemake_interface_logger_plugins.wrapper import LoggerWrapper

• CommonLoggerSettings

  from snakemake_interface_logger_plugins.settings import CommonLoggerSettings

Quickstart

This quickstart demonstrates how to create a basic Snakemake logger plugin. It defines `MyLoggerPlugin` (the entry point for Snakemake) and `MyLoggerWrapper` (which handles the actual logging messages). It also shows how to register custom command-line arguments and access them via settings. To use, save the code as `my_custom_logger.py` and run Snakemake with `--logger-plugin my_custom_logger.py`.

import os
from argparse import ArgumentParser
from snakemake_interface_logger_plugins.plugin import LoggerPlugin
from snakemake_interface_logger_plugins.wrapper import LoggerWrapper
from snakemake_interface_logger_plugins.settings import CommonLoggerSettings

# Save this as my_custom_logger.py

class MyLoggerPlugin(LoggerPlugin):
    """A simple custom logger plugin for Snakemake."""
    def __init__(self, settings: CommonLoggerSettings | None = None):
        self.settings = settings

    def register_args(self, argparser: ArgumentParser):
        """Register command line arguments for this plugin."""
        argparser.add_argument(
            "--my-logger-prefix",
            default="[CUSTOM LOG]",
            help="Prefix for custom logger messages."
        )

    def setup(self):
        """Perform setup actions before any logging occurs."""
        # Access custom args via self.settings if defined
        # print(f"{self.settings.my_logger_prefix} Initializing MyLoggerPlugin...")
        pass # For this minimal example, we don't need complex setup

    def get_wrapper(self) -> LoggerWrapper:
        """Return an instance of the logger wrapper."""
        return MyLoggerWrapper(self.settings)

class MyLoggerWrapper(LoggerWrapper):
    """Handles actual logging events."""
    def __init__(self, settings: CommonLoggerSettings | None = None):
        self.settings = settings

    def handle_info(self, msg: str):
        prefix = getattr(self.settings, 'my_logger_prefix', '[MYLOGGER]')
        print(f"{prefix} INFO: {msg}")

    def handle_error(self, msg: str):
        prefix = getattr(self.settings, 'my_logger_prefix', '[MYLOGGER]')
        print(f"{prefix} ERROR: {msg}")

    # Implement other handle_* methods as needed for full functionality
    # e.g., handle_log, handle_job_info, handle_job_error, handle_start, handle_finish, etc.
    # For this example, we only show info and error.
    # All abstract methods must be implemented, even if with a 'pass'
    def handle_log(self, msg: str):
        prefix = getattr(self.settings, 'my_logger_prefix', '[MYLOGGER]')
        print(f"{prefix} LOG: {msg}")

    def handle_start(self):
        prefix = getattr(self.settings, 'my_logger_prefix', '[MYLOGGER]')
        print(f"{prefix} Workflow started.")

    def handle_finish(self):
        prefix = getattr(self.settings, 'my_logger_prefix', '[MYLOGGER]')
        print(f"{prefix} Workflow finished.")

    def handle_job_info(self, job_info: dict):
        prefix = getattr(self.settings, 'my_logger_prefix', '[MYLOGGER]')
        print(f"{prefix} Job info: {job_info.get('jobid')} - {job_info.get('output')}")

    def handle_job_error(self, job_info: dict):
        prefix = getattr(self.settings, 'my_logger_prefix', '[MYLOGGER]')
        print(f"{prefix} Job error: {job_info.get('jobid')} - {job_info.get('output')}")

# To run this with Snakemake, save it as `my_custom_logger.py` and then run:
# snakemake --snakefile Snakefile --logger-plugin my_custom_logger.py --my-logger-prefix "[SNAKEMAKE-CUSTOM]" -c1
# (Requires a simple Snakefile, e.g., 'rule all: run: print("Hello from Snakemake")')