✍️ Writing Your Own Plugins¶

This tutorial explains how to extend Meerschaum with plugins. For general information, consult the Types of Plugins reference page.

Meerschaum's plugin system is designed to be simple so you can get your plugins working quickly. Plugins are Python packages defined in the Meerschaum configuration plugins directory and are imported at startup under the global namespace plugins.

To create your plugin, follow these steps:

1. Navigate to your Meerschaum plugins directory.

   Linux / MacOSWindowsMeerschaum Compose

   ~/.config/meerschaum/plugins

   %APPDATA%\Meerschaum\plugins

   plugins/

   If you have set MRSM_ROOT_DIR or MRSM_PLUGINS_DIR, navigate to your designated plugins directory.

2. Create your Python module.
   Create either <name>.py or <name>/__init__.py.

   Plugins may have hyphenated names, e.g. mongodb-connector:

   

3. (Optional) Define your plugin's __version__ string.

   Plugin repositories need __version__

   To publish changes to a repository with register plugin, you must increment __version__ (according to SemVer). You may omit __version__ for the initial release, but subsequent releases need the version defined.

   # ~/.config/meerschaum/plugins/example.py
   __version__ = '0.0.1'
   

4. (Optional) Define your required dependencies.

   Your plugin will be run from a virtual environment and therefore may not be able to import packages that aren't declared.

   These packagess will be installed into the plugin's virtual environment at installation or with mrsm install required <plugins>.

   Depend on other plugins

   Dependencies that start with plugin: are Meerschaum plugins. To require a plugin from another repository, add add the repository's keys after the plugin name (e.g. plugin:foo@api:bar).

   required = ['rich>=9.8.0', 'pandas', 'plugin:foo']
   

5. Write your functions.

   Special functions are fetch(), sync(), register(), setup(), and <package_name>(), and you can use the @make_action decorator to designate additional functions as actions. Below is more information on these functions.

   Don't forget keyword arguments!

   You must include a **kwargs argument to capture optional arguments. The functions fetch(), sync(), and register() also require a pipe positional argument for the meerschaum.Pipe object being synced.

   You may find the supplied arguments useful for your implementation (e.g. begin and end datetime.datetime objects). Run mrsm -h or mrsm show help to see the available keyword arguments.

### BAD - DON'T DO THIS
import pandas as pd
def fetch(pipe, **kw):
    return pd.read_csv('data.csv')

### GOOD - DO THIS INSTEAD
def fetch(pipe, **kw):
    import pandas as pd
    return pd.read_csv('data.csv')

Functions¶

Plugins are just modules with functions. This section explains the roles of the following special functions:

• register(pipe: mrsm.Pipe, **kwargs)
  Return a pipe's initial parameters dictionary.
• fetch(pipe: mrsm.Pipe, **kwargs)
  Return a DataFrame, list of dictionaries, or generator of DataFrame-like chunks.
• sync(pipe: mrsm.Pipe, **kwargs)
  Override a pipe's sync process for finer-grained control.
• @make_connector
  Create new connector types.
• @make_action
  Create new commands.
• @api_plugin
  Create new FastAPI endpoints.
• @dash_plugin and @web_page
  Create new Dash pages on the Web Console.
• @pre_sync_hook and @post_sync_hook
  Inject callbacks when pipes are synced by the sync pipes action.
• setup(**kwargs)
  Executed during plugin installation or with mrsm setup plugins <plugin>.

The register() Function¶

The register() function returns a new pipe's parameters dictionary.

The below example is register() from the noaa plugin:

def register(pipe: 'meerschaum.Pipe') -> Dict[str, Any]:
    """
    Prompt the user for stations when registering new pipes.
    """
    stations = ask_for_stations(pipe)
    return {
        'columns': {'datetime': 'timestamp', 'id': 'station',},
        'noaa': {'stations': stations,},
    }

The fetch() Function¶

The fastest way to leverage Meerschaum's syncing engine is with the fetch() function. Simply return a DataFrame (or list of dictionaries) or a chunk generator.

Below is an example of a simple fetch plugin:

### ~/.config/meerschaum/plugins/example.py

__version__ = '0.0.1'
required = []

import random
from datetime import datetime
import meerschaum as mrsm

def register(pipe: mrsm.Pipe, **kw):
    return {
        'columns': {
            'datetime': 'dt',
            'id': 'id',
        }
    }

def fetch(pipe: mrsm.Pipe, **kw):
    return [{
        'dt': datetime.utcnow(),
        'id': 1,
        'val': random.randint(0, 100),
    }]

If your plugin will be fetching large amounts of data, you may return a generator of DataFrame-like chunks:

def fetch(pipe, **kw) -> Generator[List[Dict[str, Any]]]:
    return (
        [
            {'id': i, 'val': 10.0 * i},
            {'id': i+1, 'val': 20.0 * i},
        ] for i in range(10)
    )

Chunking handles any iterable, so you may return a simple generator or yield the chunks yourself.

def fetch(pipe: mrsm.Pipe, **kw) -> Iterator['pd.DataFrame']:
    import pandas as pd
    return pd.read_csv('very-large.csv', chunksize=10_000)

def fetch(pipe: mrsm.Pipe, **kw) -> Iterator['pd.DataFrame']:
    import pandas as pd
    for file_name in ['a.csv', 'b.csv', 'c.csv']:
        yield pd.read_csv(file_name)

Robust Fetch Patterns¶

Real-world data sources fail intermittently, rate-limit, and page their responses. The patterns below keep a fetch() resilient. They're plain Python — Meerschaum doesn't impose a retry helper — so adapt them to your source.

Retry with exponential backoff. Wrap transient failures (timeouts, 5xx responses) in a bounded retry loop. Raising from fetch() lets Meerschaum mark the sync as failed; only raise once retries are exhausted.

def fetch(pipe: mrsm.Pipe, **kw):
    import time
    import requests

    url = "https://api.example.com/data"
    last_exc = None
    for attempt in range(5):
        try:
            response = requests.get(url, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            last_exc = e
            time.sleep(2 ** attempt)  ### 1s, 2s, 4s, 8s, 16s
    raise RuntimeError(f"Failed to fetch from {url} after retries: {last_exc}")

Respect rate limits. When the API signals throttling (commonly HTTP 429 with a Retry-After header), wait the indicated time instead of failing.

def fetch(pipe: mrsm.Pipe, **kw):
    import time
    import requests

    url = "https://api.example.com/data"
    while True:
        response = requests.get(url, timeout=30)
        if response.status_code == 429:
            wait_s = int(response.headers.get('Retry-After', 5))
            time.sleep(wait_s)
            continue
        response.raise_for_status()
        return response.json()

Stream paginated results as a generator. For large or paged endpoints, yield each page so Meerschaum syncs chunk-by-chunk and memory stays flat. Use begin/end to fetch only the needed window for incremental syncs.

def fetch(pipe: mrsm.Pipe, begin=None, end=None, **kw):
    import requests

    url = "https://api.example.com/data"
    params = {'page': 1}
    if begin is not None:
        params['begin'] = begin.isoformat()
    if end is not None:
        params['end'] = end.isoformat()

    while True:
        response = requests.get(url, params=params, timeout=30)
        response.raise_for_status()
        payload = response.json()
        rows = payload.get('rows', [])
        if not rows:
            break
        yield rows  ### a chunk (list of dicts); Meerschaum syncs it before the next page
        params['page'] += 1

The sync() Function¶

The sync() function makes sync pipes override the built-in syncing process and behaves more like an action, returning only a SuccessTuple (e.g. True, "Success").

Sync plugins allow for much more flexibility than fetch plugins, so what you come up with may differ from the following example. In any case, below is a simple sync plugin.

### ~/.config/meerschaum/plugins/example.py

from __future__ import annotations
from typing import Tuple, Any, Optional, Dict

__version__ = '0.0.1'

required = ['requests', 'pandas']

def sync(
        pipe: meerschaum.Pipe,
        begin: Optional[datetime.datetime] = None,
        end: Optional[datetime.datetime] = None,
        params: Dict[str, Any] = None,
        **kw: Any
    ) -> Tuple[bool, str]:
    """
    This example `sync` plugin syncs multiple pipes.

    Parameters
    ----------
    pipe: meerschaum.Pipe
       The pipe to be synced.

    begin: Optional[datetime.datetime], default None
        The datetime to start searching for data (specified with `--begin`).

    end: Optional[datetime.datetime], default None
        The datetime to stop searching for data (specified with `--end`).

    kw: Any
        Additional keyword arguments you might find useful.

    Returns
    -------
    A tuple in the form (success [bool], message [str]).
    """

    ### Get data from somewhere. You decide how!
    import pandas as pd
    import requests
    import meerschaum as mrsm

    url = "https://api.example.com/json"
    params = params or {}
    if begin is not None:
        params['begin'] = begin.isoformat()
    if end is not None:
        params['end'] = end.isoformat()

    try:
        df = pd.read_json(requests.get(url, params=params).text)
    except Exception as e:
        df = None

    if df is None:
        return False, f"Failed to sync data from {url} with params: {params}"

    success, msg = pipe.sync(df)
    if not success:
        return success, msg

    another_pipe = mrsm.Pipe('foo', 'bar', instance='sql:local')
    return another_pipe.sync(df)

The @make_connector Decorator¶

Defining a new type of connector is easy:

1. Create a new class that inherits from meerschaum.connectors.Connector.
2. Decorate the class with @make_connector.
3. Define the class-level list REQUIRED_ATTRIBUTES.
4. Add the method fetch(pipe, **kwargs) that returns data.

For example, the following creates a new connector of type foo:

# plugins/foo.py

from datetime import datetime
from typing import List, Dict, Any, Optional
import meerschaum as mrsm
from meerschaum.connectors import make_connector, Connector

required = ['requests']

@make_connector
class FooConnector(Connector):

    REQUIRED_ATTRIBUTES = ['username', 'password']

    def fetch(
        self,
        pipe: mrsm.Pipe, 
        begin: Optional[datetime] = None,
        end: Optional[datetime] = None,
        **kwargs: Any
    ) -> List[Dict[str, Any]]:
        """
        Make the request to foo.com and return the response.
        """
        params = {}
        if begin:
            params['begin'] = begin.isoformat()
        if end:
            params['end'] = end.isoformat()

        response = self.session.get("https://foo.com/data", params=params)
        return response.json()

    @property
    def session(self) -> 'requests.Session':
        """
        Return a persistent session object.

        Note that required attributes (username, password)
        are ensured to be set.
        """
        _sesh = self.__dict__.get('_session', None)
        if _sesh is not None:
            return _sesh

        import requests
        self._session = requests.Session()
        self._session.auth = (self.username, self.password)
        return self._session

You can register new foo connectors via mrsm bootstrap connector, which would prompt the user for each attribute in REQUIRED_ATTRIBUTES (username, password).

You may also define your new foo connectors as environment variables, e.g.:

export MRSM_FOO_BAR='{
    "username": "bar",
    "password": "fuzz"
}'
export MRSM_FOO_BAZ='{
    "username": "baz",
    "password": "fizz"
}'

The @make_action Decorator¶

Your plugin may extend Meerschaum by providing additional actions. Actions are regular Python functions but with perks:

1. Actions are integrated into the larger Meerschaum system.
   Actions from plugins may be executed as background jobs, via the web console, or via the API, just like standard actions.
2. Actions inherit the standard Meerschaum keyword arguments.
   You can use flags such as --begin, --end, etc., or even add your own custom flags.

### ~/.config/meerschaum/plugins/sing.py

from meerschaum.plugins import make_action

@make_action
def sing_song(**kw):
    return True, "This action is called 'sing song'."

@make_action(daemon=False)
def sing_tune(**kw):
    return True, "The action `sing tune` will not be executed in the CLI daemon context."

def sing(**kw):
    """
    Functions with the same name as the plugin are considered actions.
    """

    ### An action returns a tuple of success and message.
    ### If the success bool is `True` and the message is 'Success',
    ### nothing will be printed.

    return True, "Hello, World!"

Suggest Auto-Completions¶

Return a list of options from a function complete_<action>(), and these options will be suggested in the Meerschaum shell. The keyword arguments passed to complete_<action>() are line, sysargs, action, and the currently parsed flags.

@make_action
def foo_bar(**kw):
    return True, "Success"

def complete_foo_bar(**kw):
    return ['option 1', 'option 2']

The @api_plugin Decorator¶

Meerschaum plugins may also extend the web API by accessing the FastAPI app. Use the @api_plugin decorator to define an initialization function that will be executed on the command mrsm start api.

The only argument for the initalization function should be app for the FastAPI application.

For your endpoints, arguments will be used as HTTP parameters, and to require that the user must be logged in, import manager from meerschaum.api and add the argument curr_user = fastapi.Depends(manager):

### ~/.config/meerschaum/plugins/example.py

from meerschaum.plugins import api_plugin

@api_plugin
def init_plugin(app):
    """
    This function is executed immediately after the `app` is initialized.
    """

    import fastapi
    from meerschaum.api import manager

    @app.get('/my/new/path')
    def new_path(
        curr_user = fastapi.Depends(manager)
    ):
        """
        This is my new API endpoint.
        """
        return {'message': f'The current user is {curr_user.name}'}

The @dash_plugin and @web_page Decorators¶

Add new Plotly Dash pages to the Web Console with @dash_plugin and @web_page.

Like @api_plugin, @dash_app designates an initialization function which accepts the Dash app. Within this function, you can create callbacks with @dash_app.callback().

Functions decorated with @web_page return a Dash layout. These don't need to reside within the @dash_app-decorated function, but this is recommended to improve lazy-loading performance.

### ~/.config/meerschaum/plugins/example.py

from meerschaum.plugins import dash_plugin, web_page

@dash_plugin
def init_dash(dash_app):

    import dash.html as html
    import dash_bootstrap_components as dbc
    from dash import Input, Output, no_update

    ### Routes to '/dash/my-page'
    @web_page('/my-page', login_required=False)
    def my_page():
        return dbc.Container([
            html.H1("Hello, World!"),
            dbc.Button("Click me", id='my-button'),
            html.Div(id="my-output-div"),
        ])

    @dash_app.callback(
        Output('my-output-div', 'children'),
        Input('my-button', 'n_clicks'),
    )
    def my_button_click(n_clicks):
        if not n_clicks:
            return no_update
        return html.P(f'You clicked {n_clicks} times!')

from meerschaum.plugins import web_page, dash_plugin

@dash_plugin
def init_dash(dash_app):

    import dash.html as html
    import dash.dcc as dcc
    from dash import Input, Output, State, no_update
    import dash_bootstrap_components as dbc

    ### Omitting arguments will default to the path '/dash/items'.
    ### `login_required` is `True` by default.
    @web_page
    def items():
        return dbc.Container([
            dcc.Location(id='my-location'),
            html.Div(id='my-output-div'),
        ])

    @dash_app.callback(
        Output('my-output-div', 'children'),
        Input('my-location', 'pathname'),
    )
    def render_page_from_url(pathname):
        if not str(pathname).startswith('/dash/items'):
            return no_update

        item_id = pathname.replace('/dash/items', '').lstrip('/').rstrip('/').split('/')[0]
        if not item_id:
            return html.P("You're looking at the base /items page.")

        return html.P(f"You're looking at item '{item_id}'.")

Styling Your Page¶

The Web Console is dark by default (the Bootswatch Darkly theme plus the dbc_dark component overrides), and your page inherits that look automatically — most plugins don't need to do anything.

If you'd rather start from a clean light theme, opt out per page with @web_page(dark_theme=False):

@web_page('/my-page', dark_theme=False)
def my_page():
    return dbc.Container([html.H1("My light-themed page")])

While that page is active, the Web Console switches the Bootstrap theme to the light Flatly build and drops the dbc_dark class — and switches back to dark for the console and every other page (including on in-app navigation). Both are Bootswatch themes, so standard dash-bootstrap-components markup looks right either way; you can layer your own CSS on top.

The @pre_sync_hook and @post_sync_hook Decorators¶

You can tap into the built-in syncing engine via the sync pipes action by decorating callback functions with @pre_sync_hook and/or @post_sync_hook. Both callbacks accept a positional pipe argument, and other useful contextual arguments as passed as keyword arguments (if your function accepts them).

Add **kwargs (optional) for additional context of sync, e.g.:

• sync_method (Callable[[], mrsm.SuccessTuple])
  A bound function of either pipe.sync(), pipe.verify(), or pipe.deduplicate().
• sync_timestamp (datetime)
  The UTC datetime captured right before the sync began. This value is the same for pre- and post-syncs.

The following keywords arguments are available only for @post_sync_hook callbacks:

• success_tuple (mrsm.SuccessTuple)
  The return value of sync_method (SuccessTuple i.e. Tuple[bool, str]) to indicate whether the sync completed successfully.
• sync_complete_timestamp (datetime)
  The UTC datetime captured right after the sync completed.
• sync_duration (float)
  The duration of the sync in seconds.

All other keyword arguments correspond to the flags used to initiate the sync. See the following useful arguments:

• name (str)
  If the sync was spawned by a background job with a --name argument, you may track the job's syncs with name.
• schedule (str)
  If you spawn your syncing jobs with -s, you may see the job's cadence with schedule.
• min_seconds (float)
  If you run continous syncs (i.e. with --loop), the value of --min-seconds will be available as min_seconds.
• begin and end (datetime | int)
  The values for --begin and --end.
• params (dict[str, Any])
  The value for --params.

Sync hooks don't need to return anything, but if a SuccessTuple is returned from your callback function, it will be pretty-printed alongside the normal success messages.

When the hooks fire and in what order¶

For each pipe synced by sync pipes, the engine:

1. Captures sync_timestamp (UTC), then runs all registered @pre_sync_hook callbacks and prints any returned SuccessTuples.
2. Runs the chosen sync_method (pipe.sync, pipe.verify, or pipe.deduplicate — verification and deduplication are selected by the --verify / --deduplicate flags).
3. Computes sync_duration and captures sync_complete_timestamp, then runs all registered @post_sync_hook callbacks and prints any returned SuccessTuples.

Because the same function may be registered as both a pre- and a post-hook, branch on whether success_tuple is None (it is only set for post-syncs) to tell the two phases apart.

from datetime import datetime
import meerschaum as mrsm
from meerschaum.plugins import pre_sync_hook

@pre_sync_hook
def announce_sync(
        pipe: mrsm.Pipe,
        sync_timestamp: datetime | None = None,
        begin=None,
        end=None,
        **kwargs
    ) -> mrsm.SuccessTuple:
    """Log the pipe and requested window right before syncing."""
    print(f"About to sync {pipe} at {sync_timestamp} (begin={begin}, end={end}).")
    return True, "Success"

from datetime import datetime
import meerschaum as mrsm
from meerschaum.plugins import post_sync_hook

@post_sync_hook
def notify_on_result(
        pipe: mrsm.Pipe,
        success_tuple: mrsm.SuccessTuple,
        sync_duration: float | None = None,
        **kwargs
    ) -> mrsm.SuccessTuple:
    """Send an alert if a sync failed."""
    success, message = success_tuple
    if not success:
        ### Replace with your real notifier (email, Slack, etc.).
        print(f"ALERT: {pipe} failed after {sync_duration:.2f}s: {message}")
    return True, f"Handled result for {pipe}."

from datetime import datetime
import meerschaum as mrsm
from meerschaum.plugins import pre_sync_hook, post_sync_hook

@pre_sync_hook
@post_sync_hook
def log_sync(
        pipe: mrsm.Pipe,
        success_tuple: mrsm.SuccessTuple | None = None,
        sync_timestamp: datetime | None = None,
        sync_complete_timestamp: datetime | None = None,
        sync_duration: float | None = None,
        **kwargs
    ) -> mrsm.SuccessTuple:
    """
    Execute this callback immediately before and after syncing.
    """
    if success_tuple is None:
        print(f"About to sync {pipe} at {sync_timestamp}...")
        return True, "Success"

    success, message = success_tuple
    prefix = "successfully" if success else "fail to"

    msg = (
        f"It took {sync_duration} seconds to {prefix} sync {pipe}\n"
        + f"({sync_timestamp} - {sync_complete_timestamp})"
    )
    return True, msg

The setup() Function¶

setup() is a one-time initialization hook. It runs after your plugin's required dependencies are installed into its virtual environment, specifically:

• when the plugin is first installed (mrsm install plugin <plugin>), and
• when you explicitly re-run it with mrsm setup plugins <plugin>.

It is not called on every sync or action — use it strictly for work that should happen once at install time. It must return a SuccessTuple (a (bool, str) tuple); a False result is reported to the user and signals that setup failed.

Typical uses:

• Verify dependencies / external tooling that required can't express (a system binary, a driver, a service being reachable).
• One-time initialization such as downloading a model, creating a working directory, or seeding default plugin config via write_plugin_config().

Below is a snippet from the apex plugin which initializes a Selenium WebDriver.

def setup(**kw):
    global geckodriver_location
    try:
        from webdriver_manager.firefox import GeckoDriverManager
        geckodriver_location = GeckoDriverManager().install()
    except Exception as e:
        return False, str(e)
    return True, "Success"

Keyword Arguments¶

There are a many useful command line arguments provided to plugins as keyword arguments. To add customizability to your plugins, consider inspecting the keys in the **kwargs dictionary.

### ~/.config/meerschaum/plugins/example.py

def example(**kwargs):
    from meerschaum.utils.formatting import pprint
    pprint(kwargs)
    return True, "Success"

Useful Command Line and Keyword Arguments¶

You can see the available command line options with the command mrsm show help or mrsm --help. Expand the table below to see more information about usefult builtin arguments.

Custom Command Line Options¶

In case the built-in command line options are not sufficient, you can add arguments with add_plugin_argument(). This function takes the same arguments as the parser.add_argument() function from argparse and will include your plugin's arguments in the --help text.

### ~/.config/meerschaum/plugins/example.py

from meerschaum.plugins import add_plugin_argument
add_plugin_argument('--foo', type=int, help="This is my help text!")

The above code snippet will produce append the following text to the --help or show help text:

Plugin 'example' options:
  --foo FOO             This is my help text!

sysargs , shell, and Other Edge Cases¶

Generally, using built-in or custom arguments mentioned above should cover almost every use case. In case you have specific needs, the arguments sysargs, sub_args, filtered_sysargs, shell, and line are provided.

Working With Plugins¶

Plugins are just Python modules, so you can write custom code and share it amongst other plugins (i.e. a library plugin).

At run time, plugins are imported under the global plugins namespace, but you'll probably be testing plugins directly when the plugins namespace isn't created. That's where Plugin objects come in handy: they contain a number of convenience functions so you can cross-pollinate between plugins.

Package Management¶

Meerschaum plugins only need to be modules ― no package metadata required. But if you plan on managing your plugins as proper packages (e.g. to publish to repositories like PyPI), simply add the entry point meerschaum.plugins to your package metadata:

from setuptools import setup

setup(
    ...,
    entry_points = {
        'meerschaum.plugins': [
            'foo = foo',
        ],
    },
)

[project.entry-points."meerschaum.plugins"]
foo = "foo"

Import Another Plugin¶

Accessing the member module of the Plugin object will import its module:

# ~/.config/meerschaum/plugins/foo.py
import meerschaum as mrsm
bar = mrsm.Plugin('bar').module

Plugins in a REPL¶

Plugins run from virtual environments, so to import your plugin in a REPL, you'll need to activate its environment before importing:

>>> import meerschaum as mrsm
>>> plugin = mrsm.Plugin('foo')
>>> plugin.activate_venv()
>>> foo = plugin.module

Plugins in Scripts¶

You can also pass a plugin to the Venv virtual environment manager, which handles activating and deactivating environments.

from meerschaum.utils.venv import Venv
from meerschaum import Plugin
plugin = Plugin('foo')
with Venv(plugin):
    foo = plugin.module