openstack/deb-python-taskflow
Code Issues Proposed changes

By default use a in memory backend (when none is provided)

Browse Source 
Change-Id: I6891d53389f302f104d45d22e489cf66feb85fd8
This commit is contained in:
Joshua Harlow
2015-03-18 16:16:09 -07:00
parent dc2fdaf6b9
commit f815480c09
2 changed files with 21 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
taskflow/persistence/path_based.py
View File

@@ -45,6 +45,8 @@ class PathBasedBackend(base.Backend):
@six.add_metaclass(abc.ABCMeta) @six.add_metaclass(abc.ABCMeta)
class PathBasedConnection(base.Connection): class PathBasedConnection(base.Connection):
"""Base class for path based backend connections."""
def __init__(self, backend): def __init__(self, backend):
self._backend = backend self._backend = backend
self._book_path = self._join_path(backend.path, "books") self._book_path = self._join_path(backend.path, "books")
@@ -187,7 +189,8 @@ class PathBasedConnection(base.Connection):
def _do_update_flow_details(self, flow_detail, transaction, def _do_update_flow_details(self, flow_detail, transaction,
ignore_missing=False): ignore_missing=False):
flow_path = self._get_obj_path(flow_detail) flow_path = self._get_obj_path(flow_detail)
self._update_object(flow_detail, transaction, ignore_missing) self._update_object(flow_detail, transaction,
ignore_missing=ignore_missing)
for atom_details in flow_detail: for atom_details in flow_detail:
atom_path = self._get_obj_path(atom_details) atom_path = self._get_obj_path(atom_details)
link_path = self._join_path(flow_path, atom_details.uuid) link_path = self._join_path(flow_path, atom_details.uuid)
@@ -198,7 +201,7 @@ class PathBasedConnection(base.Connection):
def update_flow_details(self, flow_detail, ignore_missing=False): def update_flow_details(self, flow_detail, ignore_missing=False):
with self._transaction() as transaction: with self._transaction() as transaction:
return self._do_update_flow_details(flow_detail, transaction, return self._do_update_flow_details(flow_detail, transaction,
ignore_missing) ignore_missing=ignore_missing)
def get_atoms_for_flow(self, flow_uuid): def get_atoms_for_flow(self, flow_uuid):
flow_path = self._join_path(self.flow_path, flow_uuid) flow_path = self._join_path(self.flow_path, flow_uuid)
@@ -213,7 +216,7 @@ class PathBasedConnection(base.Connection):
def update_atom_details(self, atom_detail, ignore_missing=False): def update_atom_details(self, atom_detail, ignore_missing=False):
with self._transaction() as transaction: with self._transaction() as transaction:
return self._update_object(atom_detail, transaction, return self._update_object(atom_detail, transaction,
ignore_missing) ignore_missing=ignore_missing)
def _do_destroy_logbook(self, book_uuid, transaction): def _do_destroy_logbook(self, book_uuid, transaction):
book_path = self._join_path(self.book_path, book_uuid) book_path = self._join_path(self.book_path, book_uuid)

20
taskflow/storage.py
View File

@@ -22,6 +22,7 @@ import six
from taskflow import exceptions from taskflow import exceptions
from taskflow import logging from taskflow import logging
from taskflow.persistence.backends import impl_memory
from taskflow.persistence import logbook from taskflow.persistence import logbook
from taskflow import retry from taskflow import retry
from taskflow import states from taskflow import states
@@ -121,6 +122,10 @@ class Storage(object):
atom_details, flow_details) for use by engines. This makes it easier to atom_details, flow_details) for use by engines. This makes it easier to
interact with the underlying storage & backend mechanism through this interact with the underlying storage & backend mechanism through this
interface rather than accessing those objects directly. interface rather than accessing those objects directly.
NOTE(harlowja): if no backend is provided then a in-memory backend will
be automatically used and the provided flow detail object will be placed
into it for the duration of this objects existence.
""" """
injector_name = '_TaskFlow_INJECTOR' injector_name = '_TaskFlow_INJECTOR'
@@ -134,6 +139,13 @@ class Storage(object):
def __init__(self, flow_detail, backend=None, scope_fetcher=None): def __init__(self, flow_detail, backend=None, scope_fetcher=None):
self._result_mappings = {} self._result_mappings = {}
self._reverse_mapping = {} self._reverse_mapping = {}
if backend is None:
# Err on the likely-hood that most people don't make there
# objects able to be deepcopyable (resources, locks and such
# can't be deepcopied)...
backend = impl_memory.MemoryBackend({'deep_copy': False})
with contextlib.closing(backend.get_connection()) as conn:
conn.update_flow_details(flow_detail, ignore_missing=True)
self._backend = backend self._backend = backend
self._flowdetail = flow_detail self._flowdetail = flow_detail
self._transients = {} self._transients = {}
@@ -169,11 +181,9 @@ class Storage(object):
dict((name, name) for name in names)) dict((name, name) for name in names))
def _with_connection(self, functor, *args, **kwargs): def _with_connection(self, functor, *args, **kwargs):
# NOTE(harlowja): Activate the given function with a backend # Run the given functor with a backend connection as its first
# connection, if a backend is provided in the first place, otherwise # argument (providing the additional positional arguments and keyword
# don't call the function. # arguments as subsequent arguments).
if self._backend is None:
return
with contextlib.closing(self._backend.get_connection()) as conn: with contextlib.closing(self._backend.get_connection()) as conn:
functor(conn, *args, **kwargs) functor(conn, *args, **kwargs)