Hooks

Overview

pubtools projects can be extended via a hook/event/callback system based on pluggy.

This system is mainly intended for use as a publish-subscribe event dispatcher and enables the following scenarios:

• When something happens in pubtools-foo, trigger some code in pubtools-bar, without pubtools-foo having a direct dependency on pubtools-bar.

• When something happens in pubtools-foo, trigger some code in the hosting service if the task is running hosted; otherwise, do nothing.

All features provided by pluggy may be used as normal. The next few sections provide a crash course on basic usage of pluggy.

Guide: hook providers

If you are implementing a pubtools project and you want to provide hooks (which can also be considered “emitting events”), you should do the following:

1. Define hookspecs

You need to specify your hooks before using them. This can be done by defining some functions with @hookspec decorator.

# Import the pluggy objects bound to 'pubtools' namespace
from pubtools.pluggy import pm, hookspec
import sys

# Define a hookspec. This is just a plain old function, typically
# with an empty implementation (only doc string).
#
# The doc string should explain the circumstances under which this
# hook is invoked.

@hookspec
def kettle_on(kettle):
    """Called immediately before a kettle is turned on."""

@hookspec
def kettle_off(kettle):
    """Called after a kettle has been turned off."""

# Add these hookspecs onto the plugin manager.
pm.add_hookspecs(sys.modules[__name__])

Be aware that:

• There is one flat namespace for hooks across all pubtools projects, so you must take care to avoid name clashes

• You must ensure that the module containing your hookspecs is imported when your library is imported. If you forget to import the module, your hooks won’t be available.

2. Invoke hooks

At the appropriate times according to the documented behavior of your hooks, invoke them via pm.hook.

from pubtools.pluggy import pm

def make_tea(...):
   kettle.fill_water()

   # notify anyone interested that we're turning the kettle on
   pm.hook.kettle_on(kettle=kettle)

   kettle.start_boil()
   kettle.wait_boil()

   # notify again
   pm.hook.kettle_off(kettle=kettle)

   # Proceed as normal
   cup.fill_water(source=kettle)
   # (...)

When a hook is called, any plugins with matching registered @hookimpls will be invoked in LIFO order. If no plugins have registered, nothing will happen.

The above example shows a pure event emitter. More advanced patterns are possible, such as making use of values returned by hooks. See the pluggy documentation for more information.

Guide: hook implementers

If you want to implement hooks (which can also be considered as “receiving events”), you need to declare hook implementations in your project and register them with the pubtools plugin manager. Note that it is not necessary for a project to belong to the pubtools project family in order to do this.

Hook implementations can be defined using the @hookimpl decorator. The function names and signatures should match exactly the names of the corresponding hookspecs.

# Import the pluggy objects bound to 'pubtools' namespace
from pubtools.pluggy import pm, hookimpl
import sys

# Define a hookimpl. Match exactly the hookspecs declared
# in the previous example.

@hookimpl
def kettle_on(kettle):
    # Kettle causes huge spike in power consumption, we'd better
    # enable the power reserves.
    # https://en.wikipedia.org/wiki/TV_pickup
    powerbank.enable()

@hookimpl
def kettle_off(kettle):
    # Don't need the extra power any more
    powerbank.disable()

# Register this module as a plugin.
pm.register(sys.modules[__name__])

Your module must be imported for your hookimpls to take effect.

If you’re unsure whether your module will be imported, you may declare entry points in the pubtools.hooks group; pubtools will enforce that all modules in this group are imported when task_context is invoked.

For example, in setup.py, one may declare:

entry_points={
  "pubtools.hooks": [
      # Left-hand side can be anything.
      # Right-hand side is the name of your module containing a
      # call to "pm.register".
      "hooks = pubtools.foo._impl.hooks",
  ]
},

Be aware that:

• Your hookimpl could be invoked by any thread. Blocking the current thread may be inappropriate. It’s best to do only small amounts of work within a hookimpl.

• If your hookimpl raises an exception, it will propagate upwards through to the hook caller.

Guide: managing context

In the above naive example, hookimpls are standalone functions within a module, with no mechanism for passing context between themselves. For instance, the above example can’t maintain a count of boiling kettles (excluding antipatterns such as mutating globals).

A more realistic approach of registering hookimpls which allows passing around some context is to register your plugins in a two-stage process:

• First, attach to some initial event allowing you to establish context. pubtools provides task_start() for this purpose.

• When that hook is invoked, set up desired context and then register additional hooks.

# Import the pluggy objects bound to 'pubtools' namespace
from pubtools.pluggy import pm, hookimpl
import sys

class KettleSpikeHandler:
    def __init__(self):
        # Do anything I like here - configure myself from
        # settings, etc.
        self.powerbank = (...)

        # We can now maintain state between hookimpls.
        self.kettlecount = 0

    # Instance methods work OK for hookimpls.

    @hookimpl
    def kettle_on(self, kettle):
        self.kettlecount += 1
        self.powerbank.enable()

    @hookimpl
    def kettle_off(self, kettle):
        self.kettlecount -= 1
        if not self.kettlecount:
            self.powerbank.disable()

@hookimpl
def task_start():
    # When task starts, we register an instance of this object
    # as an additional plugin. This allows the object to keep
    # its own state between calls to different hookimpls.
    pm.register(KettleSpikeHandler())

# Register this module as a plugin.
# This will only register the top-level 'task_start' hookimpl.
pm.register(sys.modules[__name__])

task_start() and task_stop() are two standard hooks intended for use in setting up or tearing down a plugin context. Note that these are only conventions and may not be used uniformly across all pubtools task libraries.

API reference

pubtools.pluggy.pm

Type:

pluggy.PluginManager

A PluginManager configured for the pubtools namespace.

pubtools.pluggy.hookspec

A hookspec decorator configured for the pubtools namespace.

pubtools.pluggy.hookimpl

A hookimpl decorator configured for the pubtools namespace.

pubtools.pluggy.task_context()

Run a block of code within a task context, ensuring task lifecycle hooks are invoked when appropriate.

This is a context manager for use within pubtools task libraries where hooks shall be provided. It can be used in a with statement, as in example:

>>> with task_context():
>>>    self.do_task()

The context manager will ensure that:

• hookspecs/hookimpls are resolved across all installed libraries.

• task_start() is invoked when the block is entered.

• task_stop() is invoked when the block is exited.

Hook reference

This section lists all known hooks defined in the pubtools namespace.

Library	Hook
pubtools	task_start()
pubtools	task_stop()
pubtools	get_cert_key_paths()
pubtools	otel_exporter()
pubtools-pulplib	pulp_repository_pre_publish()
pubtools-pulplib	pulp_repository_published()
pubtools-pulp	task_pulp_flush()
pubtools-pulp	pulp_item_push_finished()
pubtools-quay	quay_repositories_cleared()
pubtools-quay	quay_repositories_removed()
pubtools-quay	quay_images_tagged()
pubtools-quay	quay_images_untagged()

task_start()

Called when a task starts.

This hook can be used to register additional hook implementations with desired context.

task_stop(failed)

Called when a task ends.

If task_start() was used to register additional hook implementations, this hook should be used to unregister them.

Parameters:

failed (bool) – True if the task is failing (i.e. exiting with non-zero exit code, or raising an exception).

get_cert_key_paths(server_url)

Get location of SSL certificates used to authenticate with a given service.

If there are multiple hook implementations and multiple values are returned, the first non-empty answer is considered canonical. The first answer is returned by the hook implementation which was registered last.

The certificates are expected to be in PEM format. It’s permitted for paths to cert and key to be the same. Callers of this hook should be prepared to receive no result, and should implement a reasonable fallback strategy in that case.

Parameters:

server_url (str) – Service URL.

Returns:

Paths to SSL certificate and key.

Return type:

(str, str)

otel_exporter()

Return an OTEL exporter, used by OTEL instrumentation.

If OTEL tracing is enabled and this hook is not implemented, a default ConsoleSpanExporter will be used.

Returns:

Instance of SpanExporter.

Return type:

opentelemetry.sdk.trace.export.SpanExporter

pulp_repository_pre_publish(repository, options)

Invoked as the first step in publishing a Pulp repository.

If a hookimpl returns a non-None value, that value will be used to replace the options for this publish. This can be used to adjust publish options from within a hook.

Args:

repository (Repository):

The repository to be published.

options (PublishOptions):

The options to use in publishing.

Returns:

options (PublishOptions):

The potentially adjusted options used for this publish.

pulp_repository_published(repository, options)

Invoked after a Pulp repository has been successfully published.

Args:

repository (Repository):

The repository which has been published.

options (PublishOptions):

The options used for this publish.

task_pulp_flush()

Invoked during task execution after successful completion of all Pulp publishes.

This hook is invoked a maximum of once per task, to indicate that all Pulp content associated with the task is considered fully up-to-date. The intended usage is to flush Pulp-derived caches or to notify systems that Pulp content may have recently changed.

pulp_item_push_finished(pulp_units, push_item)

Invoked during push tasks after each item has been processed fully.

By the time this hook is invoked, the referenced item and unit is expected to be fully uploaded into Pulp and published onto the CDN.

Args:

pulp_units (list[Unit])

A list of zero or more Pulp unit(s) created/updated for this item. Note that this information may not be available for every content type, and may only contain a subset of the Pulp fields.

push_item (PushItem)

The item which has been pushed.

quay_repositories_cleared(repository_ids)

Invoked after repositories have been cleared on Quay.

Parameters:

repository_ids (list[str]) – ID of each cleared repository.

quay_repositories_removed(repository_ids)

Invoked after repositories have been removed from Quay.

Parameters:

repository_ids (list[str]) – ID of each removed repository.

quay_images_tagged(source_ref, dest_refs)

Invoked after tagging image(s) on Quay.

Parameters:

• source_ref (str) – Source image reference.

• dest_refs (list[str]) – Destination image reference(s).

quay_images_untagged(untag_refs, lost_refs)

Invoked after untagging image(s) on Quay.

Parameters:

• untag_refs (list[str]) – Image references for which tags were removed.

• lost_refs (list[str]) – Image references (by digest) which are no longer reachable from any tag due to the untag operation.