rope_py3k / rope / base / taskhandle.py

import warnings

from rope.base import exceptions


class TaskHandle(object):

    def __init__(self, name='Task', interrupts=True):
        """Construct a TaskHandle

        If `interrupts` is `False` the task won't be interrupted by
        calling `TaskHandle.stop()`.

        """
        self.name = name
        self.interrupts = interrupts
        self.stopped = False
        self.job_sets = []
        self.observers = []

    def stop(self):
        """Interrupts the refactoring"""
        if self.interrupts:
            self.stopped = True
            self._inform_observers()

    def current_jobset(self):
        """Return the current `JobSet`"""
        if self.job_sets:
            return self.job_sets[-1]

    def add_observer(self, observer):
        """Register an observer for this task handle

        The observer is notified whenever the task is stopped or
        a job gets finished.

        """
        self.observers.append(observer)

    def is_stopped(self):
        return self.stopped

    def get_jobsets(self):
        return self.job_sets

    def create_jobset(self, name='JobSet', count=None):
        result = JobSet(self, name=name, count=count)
        self.job_sets.append(result)
        self._inform_observers()
        return result

    def _inform_observers(self):
        for observer in list(self.observers):
            observer()


class JobSet(object):

    def __init__(self, handle, name, count):
        self.handle = handle
        self.name = name
        self.count = count
        self.done = 0
        self.job_name = None

    def started_job(self, name):
        self.check_status()
        self.job_name = name
        self.handle._inform_observers()

    def finished_job(self):
        self.check_status()
        self.done += 1
        self.handle._inform_observers()
        self.job_name = None

    def check_status(self):
        if self.handle.is_stopped():
            raise exceptions.InterruptedTaskError()

    def get_active_job_name(self):
        return self.job_name

    def get_percent_done(self):
        if self.count is not None and self.count > 0:
            percent = self.done * 100 // self.count
            return min(percent, 100)

    def get_name(self):
        return self.name


class NullTaskHandle(object):

    def __init__(self):
        pass

    def is_stopped(self):
        return False

    def stop(self):
        pass

    def create_jobset(self, *args, **kwds):
        return NullJobSet()

    def get_jobsets(self):
        return []

    def add_observer(self, observer):
        pass


class NullJobSet(object):

    def started_job(self, name):
        pass

    def finished_job(self):
        pass

    def check_status(self):
        pass

    def get_active_job_name(self):
        pass

    def get_percent_done(self):
        pass

    def get_name(self):
        pass
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.