Source code for firexapp.submit.tracking_service

from abc import ABC, abstractmethod
from collections import OrderedDict

TRACKING_SERVICE_ENTRY_POINT = 'firex_tracking_service'

_services = None


#
# In case there are duplicate modules found, only keep one for each
#   (name, module_name, object_name) tuple. This prevents duplicate
#   arg registration failures when the sys.path causes the same service
#   to be found twice.
#
[docs]def prune_duplicate_module_entry_points(entry_points): id_to_entry_points = OrderedDict() for e in entry_points: key = (e.name, e.module_name, e.object_name) if key not in id_to_entry_points: id_to_entry_points[key] = e # Replace the currently stored entry point for this key if the distro is None. elif id_to_entry_points[key].distro is None and e.distro is not None: id_to_entry_points[key] = e return list(id_to_entry_points.values())
[docs]def get_tracking_services() -> (): global _services if _services is None: import entrypoints entry_pts = [entry_point for entry_point in entrypoints.get_group_all(TRACKING_SERVICE_ENTRY_POINT)] entry_pts = prune_duplicate_module_entry_points(entry_pts) entry_objects = [e.load() for e in entry_pts] _services = tuple([point() for point in entry_objects]) return _services
[docs]class TrackingService(ABC):
[docs] def extra_cli_arguments(self, arg_parser): pass
[docs] @abstractmethod def start(self, args, **kwargs) -> {}: pass
[docs] def ready_for_tasks(self, **kwargs) -> bool: return True
[docs] def ready_release_console(self, **kwargs) -> bool: return True
[docs]def get_service_name(service: TrackingService) -> str: return service.__class__.__name__