plugin import stages

beets/importer.py

@@ -720,6 +720,18 @@ def apply_choices(config):
                 for item in items:
                     config.lib.add(item)
 
+def plugin_stage(config, func):
+    """A coroutine (pipeline stage) that calls the given function with
+    each non-skipped import task. These stages occur between applying
+    metadata changes and moving/copying/writing files.
+    """
+    task = None
+    while True:
+        task = yield task
+        if task.should_skip():
+            continue
+        func(config, task)
+
 def manipulate_files(config):
     """A coroutine (pipeline stage) that performs necessary file
     manipulations *after* items have been added to the library.
@@ -913,7 +925,10 @@ def run_import(**kwargs):
         else:
             # When not autotagging, just display progress.
             stages += [show_progress(config)]
-    stages += [apply_choices(config), manipulate_files(config)]
+    stages += [apply_choices(config)]
+    for stage_func in plugins.import_stages():
+        stages.append(plugin_stage(config, stage_func))
+    stages += [manipulate_files(config)]
     if config.art:
         stages += [fetch_art(config)]
     stages += [finalize(config)]

beets/plugins.py

@@ -43,6 +43,7 @@ class BeetsPlugin(object):
         override this method.
         """
         _add_media_fields(self.item_fields())
+        self.import_stages = []
 
     def commands(self):
         """Should return a list of beets.ui.Subcommand objects for
@@ -269,6 +270,14 @@ def _add_media_fields(fields):
     for key, value in fields.iteritems():
         setattr(mediafile.MediaFile, key, value)
 
+def import_stages():
+    """Get a list of import stage functions defined by plugins."""
+    stages = []
+    for plugin in find_plugins():
+        if hasattr(plugin, 'import_stages'):
+            stages += plugin.import_stages
+    return stages
+
 
 # Event dispatch.
 

docs/changelog.rst

@@ -24,6 +24,8 @@ Changelog
   multithreaded database access could cause an internal error (with the message
   "database is locked"). This release synchronizes access to the database to
   avoid internal SQLite contention, which should avoid this error.
+* Plugins can now add parallel stages to the import pipeline. See
+  :ref:`writing-plugins`.
 * New plugin event: ``import_task_choice`` is called after an import task has an
   action assigned.
 * New plugin event: ``library_opened`` is called when beets starts up and

docs/plugins/writing.rst

@@ -276,3 +276,30 @@ are not extended, so the fields are second-class citizens. This may change
 eventually.
 
 .. _MediaFile: https://github.com/sampsyo/beets/wiki/MediaFile
+
+Add Import Pipeline Stages
+^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Many plugins need to add high-latency operations to the import workflow. For
+example, a plugin that fetches lyrics from the Web would, ideally, not block the
+progress of the rest of the importer. Beets allows plugins to add stages to the
+parallel import pipeline.
+
+Each stage is run in its own thread. Plugin stages run after metadata changes
+have been applied to a unit of music (album or track) and before file
+manipulation has occurred (copying and moving files, writing tags to disk).
+Multiple stages run in parallel but each stage processes only one task at a time
+and each task is processed by only one stage at a time.
+
+Plugins provide stages as functions that take two arguments: ``config`` and
+``task``, which are ``ImportConfig`` and ``ImportTask`` objects (both defined in
+``beets.importer``). Add such a function to the plugin's ``import_stages`` field
+to register it::
+
+    from beets.plugins import BeetsPlugin
+    class ExamplePlugin(BeetsPlugin):
+        def __init__(self):
+            super(ExamplePlugin, self).__init__()
+            self.import_stages = [self.stage]
+        def stage(self, config, task):
+            print('Importing something!')