Skip to content

Instantly share code, notes, and snippets.

@dir01
Created March 17, 2012 09:02
Show Gist options
  • Save dir01/2056881 to your computer and use it in GitHub Desktop.
Save dir01/2056881 to your computer and use it in GitHub Desktop.
class StepsProcessor(object):
def __init__(self, steps, filename):
self.steps = steps
self.filename = filename
def process(self):
for i, step in self.enumerate_steps_to_process():
try:
self.process_step(step)
except Exception:
self.report_exception_on_step(step)
def report_exception_on_step(self, step):
traceback.print_exc()
print 'Exception occured on step #%s, "%s".' % (
self.get_id_by_step(step), self.get_step_name(step)
)
next_step = self.get_next_step_by_step(step)
print 'Next step is #%s, "%s."' % (
self.get_id_by_step(next_step), self.get_step_name(step)
)
sys.exit()
def get_steps_to_process(self):
return self.steps[self.get_first_step_to_process():]
def enumerate_steps_to_process(self):
return enumerate(self.get_steps_to_process())
def get_first_step_to_process(self):
step = self.get_saved_step_id()
if not step:
return 0
elif self.is_user_agreed_to_resume_from_step(step):
return step
else:
return 0
def is_user_agreed_to_resume_from_step(self, step):
while True:
try:
return asbool(raw_input('Resume from step %s?: ' % self.get_id_by_step(step)))
except ValueError:
pass
def process_step(self, step):
self.save_step_to_state(step)
step()
def save_step_to_state(self, step):
self.save_step_id(self.get_id_by_step(step))
def get_next_step_by_step(self, step):
next_step_id = self.get_id_by_step(step) + 1
return self.get_step_by_id(next_step_id)
def get_id_by_step(self, step):
return self.steps.index(step)
def get_step_by_id(self, step_id):
return self.steps[step_id]
def get_step_name(self, step):
if isinstance(step, functools.partial):
return step.func.__name__
return step.__name__
def save_step_id(self, step_id):
with self.open_state_file() as state_file:
state_file.write(str(step_id))
def get_saved_step_id(self):
with self.open_state_file() as state_file:
try:
return int(state_file.read())
except (IOError, ValueError):
return None
def open_state_file(self):
return open(self.filename, 'wb')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment