sqlalchemy.engine.interfaces.CreateEnginePlugin

A set of hooks intended to augment the construction of an _engine.Engine object based on entrypoint names in a URL.

The purpose of _engine.CreateEnginePlugin is to allow third-party systems to apply engine, pool and dialect level event listeners without the need for the target application to be modified; instead, the plugin names can be added to the database URL. Target applications for _engine.CreateEnginePlugin include:

• connection and SQL performance tools, e.g. which use events to track number of checkouts and/or time spent with statements
• connectivity plugins such as proxies

A rudimentary _engine.CreateEnginePlugin that attaches a logger to an _engine.Engine object might look like:

import logging

from sqlalchemy.engine import CreateEnginePlugin
from sqlalchemy import event

class LogCursorEventsPlugin(CreateEnginePlugin):
    def __init__(self, url, kwargs):
        # consume the parameter "log_cursor_logging_name" from the
        # URL query
        logging_name = url.query.get("log_cursor_logging_name", "log_cursor")

        self.log = logging.getLogger(logging_name)

    def update_url(self, url):
        "update the URL to one that no longer includes our parameters"
        return url.difference_update_query(["log_cursor_logging_name"])

    def engine_created(self, engine):
        "attach an event listener after the new Engine is constructed"
        event.listen(engine, "before_cursor_execute", self._log_event)


    def _log_event(
        self,
        conn,
        cursor,
        statement,
        parameters,
        context,
        executemany):

        self.log.info("Plugin logged cursor event: %s", statement)

Plugins are registered using entry points in a similar way as that of dialects:

entry_points={
    'sqlalchemy.plugins': [
        'log_cursor_plugin = myapp.plugins:LogCursorEventsPlugin'
    ]

A plugin that uses the above names would be invoked from a database URL as in:

from sqlalchemy import create_engine

engine = create_engine(
    "mysql+pymysql://scott:tiger@localhost/test?"
    "plugin=log_cursor_plugin&log_cursor_logging_name=mylogger"
)

The plugin URL parameter supports multiple instances, so that a URL may specify multiple plugins; they are loaded in the order stated in the URL:

engine = create_engine(
  "mysql+pymysql://scott:tiger@localhost/test?"
  "plugin=plugin_one&plugin=plugin_twp&plugin=plugin_three")

The plugin names may also be passed directly to _sa.create_engine using the :paramref:`_sa.create_engine.plugins` argument:

engine = create_engine(
  "mysql+pymysql://scott:tiger@localhost/test",
  plugins=["myplugin"])

A plugin may consume plugin-specific arguments from the _engine.URL object as well as the kwargs dictionary, which is the dictionary of arguments passed to the _sa.create_engine call. "Consuming" these arguments includes that they must be removed when the plugin initializes, so that the arguments are not passed along to the _engine.Dialect constructor, where they will raise an _exc.ArgumentError because they are not known by the dialect.

As of version 1.4 of SQLAlchemy, arguments should continue to be consumed from the kwargs dictionary directly, by removing the values with a method such as dict.pop. Arguments from the _engine.URL object should be consumed by implementing the _engine.CreateEnginePlugin.update_url method, returning a new copy of the _engine.URL with plugin-specific parameters removed:

class MyPlugin(CreateEnginePlugin):
    def __init__(self, url, kwargs):
        self.my_argument_one = url.query['my_argument_one']
        self.my_argument_two = url.query['my_argument_two']
        self.my_argument_three = kwargs.pop('my_argument_three', None)

    def update_url(self, url):
        return url.difference_update_query(
            ["my_argument_one", "my_argument_two"]
        )

Arguments like those illustrated above would be consumed from a _sa.create_engine call such as:

from sqlalchemy import create_engine

engine = create_engine(
  "mysql+pymysql://scott:tiger@localhost/test?"
  "plugin=myplugin&my_argument_one=foo&my_argument_two=bar",
  my_argument_three='bat'
)

class MyPlugin(CreateEnginePlugin):
    def __init__(self, url, kwargs):
        if hasattr(CreateEnginePlugin, "update_url"):
            # detect the 1.4 API
            self.my_argument_one = url.query['my_argument_one']
            self.my_argument_two = url.query['my_argument_two']
        else:
            # detect the 1.3 and earlier API - mutate the
            # URL directly
            self.my_argument_one = url.query.pop('my_argument_one')
            self.my_argument_two = url.query.pop('my_argument_two')

        self.my_argument_three = kwargs.pop('my_argument_three', None)

    def update_url(self, url):
        # this method is only called in the 1.4 version
        return url.difference_update_query(
            ["my_argument_one", "my_argument_two"]
        )

When the engine creation process completes and produces the _engine.Engine object, it is again passed to the plugin via the _engine.CreateEnginePlugin.engine_created hook. In this hook, additional changes can be made to the engine, most typically involving setup of events (e.g. those defined in :ref:`core_event_toplevel`).