import logging
import os
from .state import WorkspaceState
from .bind import Bind
from .action_status import ActionStatus
def _get_logger():
return logging.getLogger(__name__)
[docs]class ActionTracker:
"""Manages and monitors actions"""
def __init__(self, action: str, state_path: str, path=None):
"""Creates a new action tracker
Attributes
----------
action: str
Name of the action to track
state_path: str
Path of the state file to store information about the action state
path: str, optional
Path of the file to check for existatnce on exit
This path is checked only if the tracker
is used as a context manager
"""
assert isinstance(action, str)
assert isinstance(state_path, str)
self.action = action
self.state_path = state_path
self.path = path
def __repr__(self):
return '<ActionTracker({})>'.format(self.action)
@property
def state(self) -> WorkspaceState:
"""WorkspaceState object linked to `self.state_path`"""
return WorkspaceState(self.state_path)
def __enter__(self):
"""Context manager enter"""
self.state[self.action] = ActionStatus.started.value
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit"""
there_is_no_exception = exc_tb is None
file_exists = self.path is None or os.path.exists(self.path)
status = (
ActionStatus.finished
if there_is_no_exception and file_exists
else ActionStatus.failed
)
if status is ActionStatus.failed:
if not there_is_no_exception:
_get_logger().info('Fail cause: {}'.format(exc_tb))
elif not file_exists:
_get_logger().info("File doesn't exist: {}".format(self.path))
self.state[self.action] = status.value
def _status_flag(self, status):
return self.status() == status.value
[docs] def started(self):
"""Returns True if the action has started and is pending"""
return self._status_flag(ActionStatus.started)
[docs] def finished(self):
"""Returns True if the action has finished"""
return self._status_flag(ActionStatus.finished)
[docs] def failed(self):
"""Returns True if the action has failed"""
return self._status_flag(ActionStatus.failed)
[docs] def undefined(self):
"""Returns True if the action state is not defined"""
return self.state[self.action] is None
[docs] def status(self):
"""Returns a string representation of the current action status"""
return self.state[self.action]