from .settings import Settings
try:
- from typing import Any, AnyStr, Callable, DefaultDict, Deque, Dict, List, Set, Tuple, Union
+ from typing import Any, AnyStr, Callable, DefaultDict, Deque, Dict, List, \
+ Optional, Set, Tuple, Union
except ImportError:
pass # For typing only
self.event = threading.Event()
self.progress = None
self.ret_value = None
- self.begin_time = None
- self.end_time = None
- self.duration = 0
+ self._begin_time: Optional[float] = None
+ self._end_time: Optional[float] = None
+ self.duration = 0.0
self.exception = None
self.logger = logging.getLogger('task')
self.lock = threading.Lock()
assert not self.running
self.executor.init(self)
self.set_progress(0, in_lock=True)
- self.begin_time = time.time()
+ self._begin_time = time.time()
self.running = True
self.executor.start()
exception = ex
with self.lock:
assert self.running, "_complete cannot be called before _run"
- self.end_time = now
+ self._end_time = now
self.ret_value = ret_value
self.exception = exception
- self.duration = now - self.begin_time # type: ignore
+ self.duration = now - self.begin_time
if not self.exception:
self.set_progress(100, True)
NotificationQueue.new_notification('cd_task_finished', self)
if not in_lock:
self.lock.release()
+ @property
+ def end_time(self) -> float:
+ assert self._end_time is not None
+ return self._end_time
+
+ @property
+ def begin_time(self) -> float:
+ assert self._begin_time is not None
+ return self._begin_time
+
def build_url(host, scheme=None, port=None):
"""