"""Processors module
This module contains classes which process application's actions to update
or replace labels within given GitHub repository. These describe both 'live'
and dry run (simulation).
"""
from .constants import DEFAULT_ERROR_RETURN, DEFAULT_SUCCESS_RETURN
from .github import GitHubError
from .printers import Printer, QuietPrinter
[docs]class RunModes:
"""Application run modes
This class describes both updating and replacing labels.
"""
@staticmethod
def _make_labels_dict(labels_spec):
return {k.lower(): (k, v) for k, v in labels_spec.items()}
[docs] @classmethod
def update_mode(cls, labels, labels_specs):
"""Application mode: Update
Creates label if it doesn't exist, updates if it does and keeps
any extra labels.
Example:
.. testcode::
import labelord
labels = { 'old': '000000', 'other': '222222' }
labels_specs = { 'new': '111111', 'old': 'FFFFFF' }
.. doctest::
>>> labelord.processors.RunModes.update_mode(labels, labels_specs)
({'new': ('new', '111111')}, {'old': ('old', 'FFFFFF')}, {})
Args:
labels (dict): Repository labels.
labels_specs (dict): New labels specifications.
Returns:
tuple(dict): Labels to create, update and delete.
"""
create = dict()
update = dict()
xlabels = cls._make_labels_dict(labels)
for name, color in labels_specs.items():
if name.lower() not in xlabels:
create[name] = (name, color)
elif name not in labels: # changed case of name
old_name = xlabels[name.lower()][0]
update[old_name] = (name, color)
elif labels[name] != color:
update[name] = (name, color)
return create, update, dict()
[docs] @classmethod
def replace_mode(cls, labels, labels_specs):
"""Application mode: Replace
Creates label if it doesn't exist, updates if it does and removes
any extra labels.
Example:
.. testcode::
import labelord
labels = { 'old': '000000', 'other': '222222' }
labels_specs = { 'new': '111111', 'old': 'FFFFFF' }
.. doctest::
>>> labelord.processors.RunModes.replace_mode(labels, labels_specs)
({'new': ('new', '111111')}, {'old': ('old', 'FFFFFF')}, {'other': ('other', '222222')})
Args:
labels (dict): Repository labels.
labels_specs (dict): New labels specifications.
Returns:
tuple(dict): Labels to create, update and delete.
"""
create, update, delete = cls.update_mode(labels, labels_specs)
delete = {n: (n, c) for n, c in labels.items()
if n not in labels_specs}
return create, update, delete
[docs]class RunProcessor:
"""Application's 'live' run
This class describes application's 'live' run - it performs an update or
a replacement (based on specified mode) of labels in given GitHub repository.
Initializing an instance will obtain GitHub and Printer class instances.
Args:
github (GitHub): GitHub class instance.
printer (Optional[BasePrinter]): Printer instance, default: None.
"""
MODES = {
'update': RunModes.update_mode,
'replace': RunModes.replace_mode
}
"""dict: Holds application run modes"""
def __init__(self, github, printer=None):
self.github = github
self.printer = printer or QuietPrinter()
def _process_generic(self, slug, key, data, event, method):
old_name, name, color = key, data[0], data[1]
try:
method(slug, name=name, color=color, old_name=old_name)
except GitHubError as error:
self.printer.event(event, Printer.RESULT_ERROR,
slug, name, color, error.code_message)
else:
self.printer.event(event, Printer.RESULT_SUCCESS,
slug, name, color)
def _process_create(self, slug, key, data):
self._process_generic(slug, key, data, Printer.EVENT_CREATE,
self.github.create_label)
def _process_update(self, slug, key, data):
self._process_generic(slug, key, data, Printer.EVENT_UPDATE,
self.github.update_label)
def _process_delete(self, slug, key, data):
self._process_generic(slug, key, data, Printer.EVENT_DELETE,
self.github.delete_label)
@staticmethod
def _process(slug, changes, processor):
for key, data in changes.items():
processor(slug, key, data)
def _run_one(self, slug, labels_specs, mode):
self.printer.add_repo(slug)
try:
labels = self.github.list_labels(slug)
except GitHubError as error:
self.printer.event(Printer.EVENT_LABELS, Printer.RESULT_ERROR,
slug, error.code_message)
else:
create, update, delete = mode(labels, labels_specs)
self._process(slug, create, self._process_create)
self._process(slug, update, self._process_update)
self._process(slug, delete, self._process_delete)
[docs] def run(self, slugs, labels_specs, mode):
"""Performs application's 'live' run
This method updates or replaces labels in given GitHub repositories and
passes information about performed operations to a printer.
Args:
slugs (list): List of GitHub repositories (in the format of User/Repo).
labels_specs (dict): New labels specifications.
mode (types.FunctionType): Application run mode (update or replace).
Returns:
int: Return code.
"""
for slug in slugs:
self._run_one(slug, labels_specs, mode)
self.printer.summary()
return (DEFAULT_ERROR_RETURN if self.printer.errors > 0
else DEFAULT_SUCCESS_RETURN)
[docs]class DryRunProcessor(RunProcessor):
"""Application's dry run
This class describes application's dry run - it simulates an update or
a replacement (based on specified mode) of labels in given GitHub repository.
Initializing an instance will obtain GitHub and Printer class instances.
Args:
github (GitHub): GitHub class instance.
printer (Optional[BasePrinter]): Printer instance, default: None.
"""
def __init__(self, github, printer=None):
super().__init__(github, printer)
def _process_create(self, slug, key, data):
self.printer.event(Printer.EVENT_CREATE, Printer.RESULT_DRY,
slug, data[0], data[1])
def _process_update(self, slug, key, data):
self.printer.event(Printer.EVENT_UPDATE, Printer.RESULT_DRY,
slug, data[0], data[1])
def _process_delete(self, slug, key, data):
self.printer.event(Printer.EVENT_DELETE, Printer.RESULT_DRY,
slug, data[0], data[1])