Continue adding docs to examples
Add more docs to the fake billing example and move the utility AttrDict to utils/misc.py at the same time (since its a useful utility to allow others to use). Change-Id: Id372032ae1545e0bd666bf991c21b21d320aba4b
This commit is contained in:
parent
faf2d155df
commit
a682717f73
@ -39,20 +39,15 @@ from taskflow.patterns import linear_flow as lf
|
|||||||
from taskflow import task
|
from taskflow import task
|
||||||
from taskflow.utils import misc
|
from taskflow.utils import misc
|
||||||
|
|
||||||
|
# INTRO: This example walks through a miniature workflow which simulates a
|
||||||
class AttrDict(object):
|
# the reception of a API request, creation of a database entry, driver
|
||||||
def __init__(self, **kwargs):
|
# activation (which invokes a 'fake' webservice) and final completion.
|
||||||
self._attrs = {}
|
#
|
||||||
for (k, v) in kwargs.items():
|
# This example also shows how a function/object (in this class the url sending)
|
||||||
if ' ' in k or k in ('self',) or not len(k):
|
# that occurs during driver activation can update the progress of a task
|
||||||
raise AttributeError("Invalid attribute name")
|
# without being aware of the internals of how to do this by associating a
|
||||||
self._attrs[k] = v
|
# callback that the url sending can update as the sending progresses from 0.0%
|
||||||
|
# complete to 100% complete.
|
||||||
def __getattr__(self, name):
|
|
||||||
try:
|
|
||||||
return self._attrs[name]
|
|
||||||
except KeyError:
|
|
||||||
raise AttributeError("No attributed named '%s'" % (name))
|
|
||||||
|
|
||||||
|
|
||||||
class DB(object):
|
class DB(object):
|
||||||
@ -63,15 +58,24 @@ class DB(object):
|
|||||||
class UrlCaller(object):
|
class UrlCaller(object):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self._send_time = 0.5
|
self._send_time = 0.5
|
||||||
|
self._chunks = 25
|
||||||
|
|
||||||
def send(self, url, data, status_cb=None):
|
def send(self, url, data, status_cb=None):
|
||||||
sleep_time = float(self._send_time) / 25
|
sleep_time = float(self._send_time) / self._chunks
|
||||||
for i in range(0, len(data)):
|
for i in range(0, len(data)):
|
||||||
time.sleep(sleep_time)
|
time.sleep(sleep_time)
|
||||||
|
# As we send the data, each chunk we 'fake' send will progress
|
||||||
|
# the sending progress that much further to 100%
|
||||||
if status_cb:
|
if status_cb:
|
||||||
status_cb(float(i) / len(data))
|
status_cb(float(i) / len(data))
|
||||||
|
|
||||||
|
|
||||||
|
# Since engines save the output of tasks to a optional persistant storage
|
||||||
|
# backend resources have to be dealt with in a slightly different manner since
|
||||||
|
# resources are transient and can not be persisted (or serialized). For tasks
|
||||||
|
# that require access to a set of resources it is a common pattern to provide
|
||||||
|
# a object (in this case this object) on construction of those tasks via the
|
||||||
|
# task constructor.
|
||||||
class ResourceFetcher(object):
|
class ResourceFetcher(object):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self._db_handle = None
|
self._db_handle = None
|
||||||
@ -126,11 +130,17 @@ class ActivateDriver(task.Task):
|
|||||||
def execute(self, parsed_request):
|
def execute(self, parsed_request):
|
||||||
print("Sending billing data to %s" % (self._url))
|
print("Sending billing data to %s" % (self._url))
|
||||||
url_sender = self._resources.url_handle
|
url_sender = self._resources.url_handle
|
||||||
|
# Note that here we attach our update_progress function (which is a
|
||||||
|
# function that the engine also 'binds' to) to the progress function
|
||||||
|
# that the url sending helper class uses. This allows the task progress
|
||||||
|
# to be tied to the url sending progress, which is very useful for
|
||||||
|
# downstream systems to be aware of what a task is doing at any time.
|
||||||
url_sender.send(self._url, json.dumps(parsed_request),
|
url_sender.send(self._url, json.dumps(parsed_request),
|
||||||
status_cb=self.update_progress)
|
status_cb=self.update_progress)
|
||||||
return self._url
|
return self._url
|
||||||
|
|
||||||
def update_progress(self, progress, **kwargs):
|
def update_progress(self, progress, **kwargs):
|
||||||
|
# Override the parent method to also print out the status.
|
||||||
super(ActivateDriver, self).update_progress(progress, **kwargs)
|
super(ActivateDriver, self).update_progress(progress, **kwargs)
|
||||||
print("%s is %0.2f%% done" % (self.name, progress * 100))
|
print("%s is %0.2f%% done" % (self.name, progress * 100))
|
||||||
|
|
||||||
@ -141,36 +151,33 @@ class DeclareSuccess(task.Task):
|
|||||||
print("All data processed and sent to %s" % (sent_to))
|
print("All data processed and sent to %s" % (sent_to))
|
||||||
|
|
||||||
|
|
||||||
SERIAL = False
|
|
||||||
if SERIAL:
|
|
||||||
engine_conf = {
|
|
||||||
'engine': 'serial',
|
|
||||||
}
|
|
||||||
else:
|
|
||||||
engine_conf = {
|
|
||||||
'engine': 'parallel',
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
# Resources (db handles and similar) of course can't be persisted so we need
|
# Resources (db handles and similar) of course can't be persisted so we need
|
||||||
# to make sure that we pass this resource fetcher to the tasks constructor so
|
# to make sure that we pass this resource fetcher to the tasks constructor so
|
||||||
# that the tasks have access to any needed resources (lazily loaded).
|
# that the tasks have access to any needed resources (the resources are
|
||||||
|
# lazily loaded so that they are only created when they are used).
|
||||||
resources = ResourceFetcher()
|
resources = ResourceFetcher()
|
||||||
flow = lf.Flow("initialize-me")
|
flow = lf.Flow("initialize-me")
|
||||||
|
|
||||||
# 1. First we extract the api request into a useable format.
|
# 1. First we extract the api request into a useable format.
|
||||||
# 2. Then we go ahead and make a database entry for our request.
|
# 2. Then we go ahead and make a database entry for our request.
|
||||||
flow.add(ExtractInputRequest(resources),
|
flow.add(ExtractInputRequest(resources), MakeDBEntry(resources))
|
||||||
MakeDBEntry(resources))
|
|
||||||
|
|
||||||
# 3. Then we activate our payment method and finally declare success
|
# 3. Then we activate our payment method and finally declare success.
|
||||||
sub_flow = gf.Flow("after-initialize")
|
sub_flow = gf.Flow("after-initialize")
|
||||||
sub_flow.add(ActivateDriver(resources), DeclareSuccess())
|
sub_flow.add(ActivateDriver(resources), DeclareSuccess())
|
||||||
flow.add(sub_flow)
|
flow.add(sub_flow)
|
||||||
|
|
||||||
|
# Initially populate the storage with the following request object,
|
||||||
|
# prepopulating this allows the tasks that dependent on the 'request' variable
|
||||||
|
# to start processing (in this case this is the ExtractInputRequest task).
|
||||||
store = {
|
store = {
|
||||||
'request': AttrDict(user="bob", id="1.35"),
|
'request': misc.AttrDict(user="bob", id="1.35"),
|
||||||
}
|
}
|
||||||
eng = engines.load(flow, engine_conf=engine_conf, store=store)
|
eng = engines.load(flow, engine_conf='serial', store=store)
|
||||||
|
|
||||||
|
# This context manager automatically adds (and automatically removes) a
|
||||||
|
# helpful set of state transition notification printing helper utilities
|
||||||
|
# that show you exactly what transitions the engine is going through
|
||||||
|
# while running the various billing related tasks.
|
||||||
with printing.PrintingListener(eng):
|
with printing.PrintingListener(eng):
|
||||||
eng.run()
|
eng.run()
|
||||||
|
@ -75,6 +75,43 @@ def get_duplicate_keys(iterable, key=None):
|
|||||||
return duplicates
|
return duplicates
|
||||||
|
|
||||||
|
|
||||||
|
class AttrDict(dict):
|
||||||
|
"""Helper utility class to create a class that can be accessed by
|
||||||
|
attribute name from a dictionary that contains a set of keys and values.
|
||||||
|
"""
|
||||||
|
@staticmethod
|
||||||
|
def _is_valid_attribute_name(name):
|
||||||
|
if not isinstance(name, six.string_types) or len(name) == 0:
|
||||||
|
return False
|
||||||
|
if name.lower().startswith('self') or name.startswith("_"):
|
||||||
|
return False
|
||||||
|
if not name[0].isalpha():
|
||||||
|
return False
|
||||||
|
for i in range(1, len(name)):
|
||||||
|
if not (name[i].isalpha() or name[i].isdigit()):
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
def __init__(self, **kwargs):
|
||||||
|
for (k, v) in kwargs.items():
|
||||||
|
if not self._is_valid_attribute_name(k):
|
||||||
|
raise AttributeError("Invalid attribute name: '%s'" % (k))
|
||||||
|
self[k] = v
|
||||||
|
|
||||||
|
def __getattr__(self, name):
|
||||||
|
if not self._is_valid_attribute_name(name):
|
||||||
|
raise AttributeError("Invalid attribute name: '%s'" % (name))
|
||||||
|
try:
|
||||||
|
return self[name]
|
||||||
|
except KeyError:
|
||||||
|
raise AttributeError("No attributed named: '%s'" % (name))
|
||||||
|
|
||||||
|
def __setattr__(self, name, value):
|
||||||
|
if not self._is_valid_attribute_name(name):
|
||||||
|
raise AttributeError("Invalid attribute name: '%s'" % (name))
|
||||||
|
self[name] = value
|
||||||
|
|
||||||
|
|
||||||
class ExponentialBackoff(object):
|
class ExponentialBackoff(object):
|
||||||
def __init__(self, attempts, exponent=2):
|
def __init__(self, attempts, exponent=2):
|
||||||
self.attempts = int(attempts)
|
self.attempts = int(attempts)
|
||||||
|
Loading…
Reference in New Issue
Block a user