openstack/taskflow
Code Issues Proposed changes

Run pyupgrade to clean up Python 2 syntaxes

Browse Source 
Update all .py source files by
 $ pyupgrade --py3-only $(git ls-files | grep ".py$")
to modernize the code according to Python 3 syntaxes.

pep8 errors are fixed by
 $ autopep8 --select=E127,E128,E501 --max-line-length 79 -r \
    --in-place taskflow

Also add the pyupgrade hook to pre-commit to avoid merging additional
Python 2 syntaxes.

Change-Id: Ifd0a0ade9789497482c7937bffd82c48acfb3d78
This commit is contained in:
Takashi Kajinami 2025-01-13 01:13:12 +09:00
parent 31f5b4e46f
commit 80e1aadc49
197 changed files with 604 additions and 1019 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.pre-commit-config.yaml
View File

@ -24,3 +24,8 @@ repos:
- id: hacking
additional_dependencies: []
exclude: '^(doc|releasenotes|tools)/.*$'
- repo: https://github.com/asottile/pyupgrade
rev: v3.18.0
hooks:
- id: pyupgrade
args: [--py3-only]

1
doc/source/conf.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");

1
releasenotes/source/conf.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2020 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");

8
taskflow/atom.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Rackspace Hosting Inc. All Rights Reserved.
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
@ -159,7 +157,7 @@ def _build_arg_mapping(atom_name, reqs, rebind_args, function, do_infer,
return required, optional
class Atom(object, metaclass=abc.ABCMeta):
class Atom(metaclass=abc.ABCMeta):
"""An unit of work that causes a flow to progress (in some manner).
An atom is a named object that operates with input data to perform
@ -379,7 +377,7 @@ class Atom(object, metaclass=abc.ABCMeta):
"""
def __str__(self):
return '"%s==%s"' % (self.name, misc.get_version_string(self))
return '"{}=={}"'.format(self.name, misc.get_version_string(self))
def __repr__(self):
return '<%s %s>' % (reflection.get_class_name(self), self)
return '<{} {}>'.format(reflection.get_class_name(self), self)

2
taskflow/conductors/backends/__init__.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/conductors/backends/impl_blocking.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@ -33,7 +31,7 @@ class BlockingConductor(impl_executor.ExecutorConductor):
persistence=None, engine=None,
engine_options=None, wait_timeout=None,
log=None, max_simultaneous_jobs=MAX_SIMULTANEOUS_JOBS):
super(BlockingConductor, self).__init__(
super().__init__(
name, jobboard,
persistence=persistence, engine=engine,
engine_options=engine_options,

9
taskflow/conductors/backends/impl_executor.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@ -104,7 +102,7 @@ class ExecutorConductor(base.Conductor, metaclass=abc.ABCMeta):
persistence=None, engine=None,
engine_options=None, wait_timeout=None,
log=None, max_simultaneous_jobs=MAX_SIMULTANEOUS_JOBS):
super(ExecutorConductor, self).__init__(
super().__init__(
name, jobboard, persistence=persistence,
engine=engine, engine_options=engine_options)
self._wait_timeout = tt.convert_to_timeout(
@ -139,8 +137,7 @@ class ExecutorConductor(base.Conductor, metaclass=abc.ABCMeta):
return not self._dead.is_set()
def _listeners_from_job(self, job, engine):
listeners = super(ExecutorConductor, self)._listeners_from_job(
job, engine)
listeners = super()._listeners_from_job(job, engine)
listeners.append(logging_listener.LoggingListener(engine,
log=self._log))
return listeners
@ -178,7 +175,7 @@ class ExecutorConductor(base.Conductor, metaclass=abc.ABCMeta):
stage_func()
self._notifier.notify("%s_end" % event_name, details)
except excp.WrappedFailure as e:
if all((f.check(*self.NO_CONSUME_EXCEPTIONS) for f in e)):
if all(f.check(*self.NO_CONSUME_EXCEPTIONS) for f in e):
consume = False
if self._log.isEnabledFor(logging.WARNING):
if consume:

4
taskflow/conductors/backends/impl_nonblocking.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@ -54,7 +52,7 @@ class NonBlockingConductor(impl_executor.ExecutorConductor):
engine_options=None, wait_timeout=None,
log=None, max_simultaneous_jobs=MAX_SIMULTANEOUS_JOBS,
executor_factory=None):
super(NonBlockingConductor, self).__init__(
super().__init__(
name, jobboard,
persistence=persistence, engine=engine,
engine_options=engine_options, wait_timeout=wait_timeout,

4
taskflow/conductors/base.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@ -25,7 +23,7 @@ from taskflow.types import notifier
from taskflow.utils import misc
class Conductor(object, metaclass=abc.ABCMeta):
class Conductor(metaclass=abc.ABCMeta):
"""Base for all conductor implementations.
Conductors act as entities which extract jobs from a jobboard, assign

2
taskflow/deciders.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

2
taskflow/engines/__init__.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/engines/action_engine/actions/base.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -19,7 +17,7 @@ import abc
from taskflow import states
class Action(object, metaclass=abc.ABCMeta):
class Action(metaclass=abc.ABCMeta):
"""An action that handles executing, state changes, ... of atoms."""
NO_RESULT = object()

4
taskflow/engines/action_engine/actions/retry.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -24,7 +22,7 @@ class RetryAction(base.Action):
"""An action that handles executing, state changes, ... of retry atoms."""
def __init__(self, storage, notifier, retry_executor):
super(RetryAction, self).__init__(storage, notifier)
super().__init__(storage, notifier)
self._retry_executor = retry_executor
def _get_retry_args(self, retry, revert=False, addons=None):

4
taskflow/engines/action_engine/actions/task.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -29,7 +27,7 @@ class TaskAction(base.Action):
"""An action that handles scheduling, state changes, ... of task atoms."""
def __init__(self, storage, notifier, task_executor):
super(TaskAction, self).__init__(storage, notifier)
super().__init__(storage, notifier)
self._task_executor = task_executor
def _is_identity_transition(self, old_state, state, task, progress=None):

6
taskflow/engines/action_engine/builder.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -58,7 +56,7 @@ TIMED_STATES = (st.ANALYZING, st.RESUMING, st.SCHEDULING, st.WAITING)
LOG = logging.getLogger(__name__)
class MachineMemory(object):
class MachineMemory:
"""State machine memory."""
def __init__(self):
@ -73,7 +71,7 @@ class MachineMemory(object):
fut.cancel()
class MachineBuilder(object):
class MachineBuilder:
"""State machine *builder* that powers the engine components.
NOTE(harlowja): the machine (states and events that will trigger

20
taskflow/engines/action_engine/compiler.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -46,12 +44,12 @@ ATOMS = (TASK, RETRY)
FLOWS = (FLOW, FLOW_END)
class Terminator(object):
class Terminator:
"""Flow terminator class."""
def __init__(self, flow):
self._flow = flow
self._name = "%s[$]" % (self._flow.name,)
self._name = "{}[$]".format(self._flow.name)
@property
def flow(self):
@ -68,7 +66,7 @@ class Terminator(object):
return '"%s[$]"' % flow_name
class Compilation(object):
class Compilation:
"""The result of a compilers ``compile()`` is this *immutable* object."""
#: Task nodes will have a ``kind`` metadata key with this value.
@ -135,7 +133,7 @@ def _add_update_edges(graph, nodes_from, nodes_to, attr_dict=None):
graph.add_edge(u, v, attr_dict=attr_dict.copy())
class TaskCompiler(object):
class TaskCompiler:
"""Non-recursive compiler of tasks."""
def compile(self, task, parent=None):
@ -147,7 +145,7 @@ class TaskCompiler(object):
return graph, node
class FlowCompiler(object):
class FlowCompiler:
"""Recursive compiler of flows."""
def __init__(self, deep_compiler_func):
@ -162,9 +160,9 @@ class FlowCompiler(object):
parent.add(tree_node)
if flow.retry is not None:
tree_node.add(tr.Node(flow.retry, kind=RETRY))
decomposed = dict(
(child, self._deep_compiler_func(child, parent=tree_node)[0])
for child in flow)
decomposed = {
child: self._deep_compiler_func(child, parent=tree_node)[0]
for child in flow}
decomposed_graphs = list(decomposed.values())
graph = gr.merge_graphs(graph, *decomposed_graphs,
overlap_detector=_overlap_occurrence_detector)
@ -223,7 +221,7 @@ class FlowCompiler(object):
return graph, tree_node
class PatternCompiler(object):
class PatternCompiler:
"""Compiles a flow pattern (or task) into a compilation unit.
Let's dive into the basic idea for how this works:

12
taskflow/engines/action_engine/completer.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -29,7 +27,7 @@ from taskflow import states as st
LOG = logging.getLogger(__name__)
class Strategy(object, metaclass=abc.ABCMeta):
class Strategy(metaclass=abc.ABCMeta):
"""Failure resolution strategy base class."""
strategy = None
@ -56,7 +54,7 @@ class RevertAndRetry(Strategy):
strategy = retry_atom.RETRY
def __init__(self, runtime, retry):
super(RevertAndRetry, self).__init__(runtime)
super().__init__(runtime)
self._retry = retry
def apply(self):
@ -73,7 +71,7 @@ class RevertAll(Strategy):
strategy = retry_atom.REVERT_ALL
def __init__(self, runtime):
super(RevertAll, self).__init__(runtime)
super().__init__(runtime)
def apply(self):
return self._runtime.reset_atoms(
@ -87,7 +85,7 @@ class Revert(Strategy):
strategy = retry_atom.REVERT
def __init__(self, runtime, atom):
super(Revert, self).__init__(runtime)
super().__init__(runtime)
self._atom = atom
def apply(self):
@ -98,7 +96,7 @@ class Revert(Strategy):
return tweaked
class Completer(object):
class Completer:
"""Completes atoms using actions to complete them."""
def __init__(self, runtime):

4
taskflow/engines/action_engine/deciders.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -26,7 +24,7 @@ from taskflow import states
LOG = logging.getLogger(__name__)
class Decider(object, metaclass=abc.ABCMeta):
class Decider(metaclass=abc.ABCMeta):
"""Base class for deciders.
Provides interface to be implemented by sub-classes.

10
taskflow/engines/action_engine/engine.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -165,7 +163,7 @@ class ActionEngine(base.Engine):
"""
def __init__(self, flow, flow_detail, backend, options):
super(ActionEngine, self).__init__(flow, flow_detail, backend, options)
super().__init__(flow, flow_detail, backend, options)
self._runtime = None
self._compiled = False
self._compilation = None
@ -474,8 +472,7 @@ class SerialActionEngine(ActionEngine):
"""Engine that runs tasks in serial manner."""
def __init__(self, flow, flow_detail, backend, options):
super(SerialActionEngine, self).__init__(flow, flow_detail,
backend, options)
super().__init__(flow, flow_detail, backend, options)
self._task_executor = executor.SerialTaskExecutor()
@ -576,8 +573,7 @@ String (case insensitive) Executor used
_default_executor_cls = executor.ParallelThreadTaskExecutor
def __init__(self, flow, flow_detail, backend, options):
super(ParallelActionEngine, self).__init__(flow, flow_detail,
backend, options)
super().__init__(flow, flow_detail, backend, options)
# This ensures that any provided executor will be validated before
# we get to far in the compilation/execution pipeline...
self._task_executor = self._fetch_task_executor(self._options)

6
taskflow/engines/action_engine/executor.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -78,7 +76,7 @@ def _revert_task(task, arguments, result, failures, progress_callback=None):
return (REVERTED, result)
class SerialRetryExecutor(object):
class SerialRetryExecutor:
"""Executes and reverts retries."""
def __init__(self):
@ -105,7 +103,7 @@ class SerialRetryExecutor(object):
return fut
class TaskExecutor(object, metaclass=abc.ABCMeta):
class TaskExecutor(metaclass=abc.ABCMeta):
"""Executes and reverts tasks.
This class takes task and its arguments and executes or reverts it.

4
taskflow/engines/action_engine/runtime.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -43,7 +41,7 @@ _EdgeDecider = collections.namedtuple('_EdgeDecider',
LOG = logging.getLogger(__name__)
class Runtime(object):
class Runtime:
"""A aggregate of runtime objects, properties, ... used during execution.
This object contains various utility methods and properties that represent

8
taskflow/engines/action_engine/scheduler.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -21,7 +19,7 @@ from taskflow import states as st
from taskflow.types import failure
class RetryScheduler(object):
class RetryScheduler:
"""Schedules retry atoms."""
def __init__(self, runtime):
@ -52,7 +50,7 @@ class RetryScheduler(object):
" intention: %s" % intention)
class TaskScheduler(object):
class TaskScheduler:
"""Schedules task atoms."""
def __init__(self, runtime):
@ -75,7 +73,7 @@ class TaskScheduler(object):
" intention: %s" % intention)
class Scheduler(object):
class Scheduler:
"""Safely schedules atoms using a runtime ``fetch_scheduler`` routine."""
def __init__(self, runtime):

8
taskflow/engines/action_engine/scopes.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -21,7 +19,7 @@ from taskflow import logging
LOG = logging.getLogger(__name__)
class ScopeWalker(object):
class ScopeWalker:
"""Walks through the scopes of a atom using a engines compilation.
NOTE(harlowja): for internal usage only.
@ -79,9 +77,9 @@ class ScopeWalker(object):
"""
graph = self._execution_graph
if self._predecessors is None:
predecessors = set(
predecessors = {
node for node in graph.bfs_predecessors_iter(self._atom)
if graph.nodes[node]['kind'] in co.ATOMS)
if graph.nodes[node]['kind'] in co.ATOMS}
self._predecessors = predecessors.copy()
else:
predecessors = self._predecessors.copy()

4
taskflow/engines/action_engine/selector.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -27,7 +25,7 @@ from taskflow.utils import iter_utils
LOG = logging.getLogger(__name__)
class Selector(object):
class Selector:
"""Selector that uses a compilation and aids in execution processes.
Its primary purpose is to get the next atoms for execution or reversion

2
taskflow/engines/action_engine/traversal.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/engines/base.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -21,7 +19,7 @@ from taskflow.types import notifier
from taskflow.utils import misc
class Engine(object, metaclass=abc.ABCMeta):
class Engine(metaclass=abc.ABCMeta):
"""Base for all engines implementations.
:ivar Engine.notifier: A notification object that will dispatch

2
taskflow/engines/helpers.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

6
taskflow/engines/worker_based/dispatcher.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -23,7 +21,7 @@ from taskflow.utils import kombu_utils as ku
LOG = logging.getLogger(__name__)
class Handler(object):
class Handler:
"""Component(s) that will be called on reception of messages."""
__slots__ = ['_process_message', '_validator']
@ -53,7 +51,7 @@ class Handler(object):
return self._validator
class TypeDispatcher(object):
class TypeDispatcher:
"""Receives messages and dispatches to type specific handlers."""
def __init__(self, type_handlers=None, requeue_filters=None):

4
taskflow/engines/worker_based/endpoint.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -19,7 +17,7 @@ from oslo_utils import reflection
from taskflow.engines.action_engine import executor
class Endpoint(object):
class Endpoint:
"""Represents a single task with execute/revert methods."""
def __init__(self, task_cls):

5
taskflow/engines/worker_based/engine.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -53,8 +51,7 @@ class WorkerBasedActionEngine(engine.ActionEngine):
"""
def __init__(self, flow, flow_detail, backend, options):
super(WorkerBasedActionEngine, self).__init__(flow, flow_detail,
backend, options)
super().__init__(flow, flow_detail, backend, options)
# This ensures that any provided executor will be validated before
# we get to far in the compilation/execution pipeline...
self._task_executor = self._fetch_task_executor(self._options,

2
taskflow/engines/worker_based/executor.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/engines/worker_based/protocol.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -147,7 +145,7 @@ def failure_to_dict(failure):
return failure.to_dict(include_args=False)
class Message(object, metaclass=abc.ABCMeta):
class Message(metaclass=abc.ABCMeta):
"""Base class for all message types."""
def __repr__(self):

6
taskflow/engines/worker_based/proxy.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -40,7 +38,7 @@ _TransportDetails = collections.namedtuple('_TransportDetails',
'driver_name', 'driver_version'])
class Proxy(object):
class Proxy:
"""A proxy processes messages from/to the named exchange.
For **internal** usage only (not for public consumption).
@ -145,7 +143,7 @@ class Proxy(object):
def _make_queue(self, routing_key, exchange, channel=None):
"""Make a named queue for the given exchange."""
queue_name = "%s_%s" % (self._exchange_name, routing_key)
queue_name = "{}_{}".format(self._exchange_name, routing_key)
return kombu.Queue(name=queue_name,
routing_key=routing_key, durable=False,
exchange=exchange, auto_delete=True,

8
taskflow/engines/worker_based/server.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -31,7 +29,7 @@ from taskflow.utils import misc
LOG = logging.getLogger(__name__)
class Server(object):
class Server:
"""Server implementation that waits for incoming tasks requests."""
def __init__(self, topic, exchange, executor, endpoints,
@ -53,8 +51,8 @@ class Server(object):
transport_options=transport_options,
retry_options=retry_options)
self._topic = topic
self._endpoints = dict([(endpoint.name, endpoint)
for endpoint in endpoints])
self._endpoints = {endpoint.name: endpoint
for endpoint in endpoints}
def _delayed_process(self, func):
"""Runs the function using the instances executor (eventually).

13
taskflow/engines/worker_based/types.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -30,7 +28,7 @@ LOG = logging.getLogger(__name__)
# TODO(harlowja): this needs to be made better, once
# https://blueprints.launchpad.net/taskflow/+spec/wbe-worker-info is finally
# implemented we can go about using that instead.
class TopicWorker(object):
class TopicWorker:
"""A (read-only) worker and its relevant information + useful methods."""
_NO_IDENTITY = object()
@ -72,14 +70,15 @@ class TopicWorker(object):
def __repr__(self):
r = reflection.get_class_name(self, fully_qualified=False)
if self.identity is not self._NO_IDENTITY:
r += "(identity=%s, tasks=%s, topic=%s)" % (self.identity,
self.tasks, self.topic)
r += "(identity={}, tasks={}, topic={})".format(
self.identity, self.tasks, self.topic)
else:
r += "(identity=*, tasks=%s, topic=%s)" % (self.tasks, self.topic)
r += "(identity=*, tasks={}, topic={})".format(
self.tasks, self.topic)
return r
class ProxyWorkerFinder(object):
class ProxyWorkerFinder:
"""Requests and receives responses about workers topic+task details."""
def __init__(self, uuid, proxy, topics,

10
taskflow/engines/worker_based/worker.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -33,7 +31,7 @@ from taskflow.utils import threading_utils as tu
LOG = logging.getLogger(__name__)
class Worker(object):
class Worker:
"""Worker that can be started on a remote host for handling tasks requests.
:param url: broker url
@ -88,13 +86,13 @@ class Worker(object):
connection_details = self._server.connection_details
transport = connection_details.transport
if transport.driver_version:
transport_driver = "%s v%s" % (transport.driver_name,
transport.driver_version)
transport_driver = "{} v{}".format(transport.driver_name,
transport.driver_version)
else:
transport_driver = transport.driver_name
try:
hostname = socket.getfqdn()
except socket.error:
except OSError:
hostname = "???"
try:
pid = os.getpid()

2
taskflow/examples/99_bottles.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

2
taskflow/examples/alphabet_soup.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/examples/build_a_car.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -111,7 +109,7 @@ def flow_watch(state, details):
def task_watch(state, details):
print('Task %s => %s' % (details.get('task_name'), state))
print('Task {} => {}'.format(details.get('task_name'), state))
flow = lf.Flow("make-auto").add(

4
taskflow/examples/buildsystem.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -56,7 +54,7 @@ class LinkTask(task.Task):
default_provides = 'executable'
def __init__(self, executable_path, *args, **kwargs):
super(LinkTask, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self._executable_path = executable_path
def execute(self, **kwargs):

4
taskflow/examples/calculate_in_parallel.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -46,7 +44,7 @@ from taskflow import task
# more uniform manner).
class Provider(task.Task):
def __init__(self, name, *args, **kwargs):
super(Provider, self).__init__(name=name, **kwargs)
super().__init__(name=name, **kwargs)
self._provide = args
def execute(self):

8
taskflow/examples/calculate_linear.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -54,7 +52,7 @@ from taskflow import task
class Provider(task.Task):
def __init__(self, name, *args, **kwargs):
super(Provider, self).__init__(name=name, **kwargs)
super().__init__(name=name, **kwargs)
self._provide = args
def execute(self):
@ -79,8 +77,8 @@ class Adder(task.Task):
# this function needs to undo if some later operation fails.
class Multiplier(task.Task):
def __init__(self, name, multiplier, provides=None, rebind=None):
super(Multiplier, self).__init__(name=name, provides=provides,
rebind=rebind)
super().__init__(name=name, provides=provides,
rebind=rebind)
self._multiplier = multiplier
def execute(self, z):

7
taskflow/examples/create_parallel_volume.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -45,7 +43,7 @@ def show_time(name):
start = time.time()
yield
end = time.time()
print(" -- %s took %0.3f seconds" % (name, end - start))
print(" -- {} took {:0.3f} seconds".format(name, end - start))
# This affects how many volumes to create and how much time to *simulate*
@ -85,8 +83,7 @@ class VolumeCreator(task.Task):
# volume create can be resumed/revert, and is much easier to use for
# audit and tracking purposes.
base_name = reflection.get_callable_name(self)
super(VolumeCreator, self).__init__(name="%s-%s" % (base_name,
volume_id))
super().__init__(name="{}-{}".format(base_name, volume_id))
self._volume_id = volume_id
def execute(self):

4
taskflow/examples/delayed_return.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -44,7 +42,7 @@ from taskflow.types import notifier
class PokeFutureListener(base.Listener):
def __init__(self, engine, future, task_name):
super(PokeFutureListener, self).__init__(
super().__init__(
engine,
task_listen_for=(notifier.Notifier.ANY,),
flow_listen_for=[])

36
taskflow/examples/distance_calculator.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -62,31 +60,24 @@ if __name__ == '__main__':
any_distance = linear_flow.Flow("origin").add(DistanceTask())
results = engines.run(any_distance)
print(results)
print("%s is near-enough to %s: %s" % (results['distance'],
0.0,
is_near(results['distance'], 0.0)))
print("{} is near-enough to {}: {}".format(
results['distance'], 0.0, is_near(results['distance'], 0.0)))
results = engines.run(any_distance, store={'a': Point(1, 1)})
print(results)
print("%s is near-enough to %s: %s" % (results['distance'],
1.4142,
is_near(results['distance'],
1.4142)))
print("{} is near-enough to {}: {}".format(
results['distance'], 1.4142, is_near(results['distance'], 1.4142)))
results = engines.run(any_distance, store={'a': Point(10, 10)})
print(results)
print("%s is near-enough to %s: %s" % (results['distance'],
14.14199,
is_near(results['distance'],
14.14199)))
print("{} is near-enough to {}: {}".format(
results['distance'], 14.14199, is_near(results['distance'], 14.14199)))
results = engines.run(any_distance,
store={'a': Point(5, 5), 'b': Point(10, 10)})
print(results)
print("%s is near-enough to %s: %s" % (results['distance'],
7.07106,
is_near(results['distance'],
7.07106)))
print("{} is near-enough to {}: {}".format(
results['distance'], 7.07106, is_near(results['distance'], 7.07106)))
# For this we use the ability to override at task creation time the
# optional arguments so that we don't need to continue to send them
@ -97,13 +88,10 @@ if __name__ == '__main__':
ten_distance.add(DistanceTask(inject={'a': Point(10, 10)}))
results = engines.run(ten_distance, store={'b': Point(10, 10)})
print(results)
print("%s is near-enough to %s: %s" % (results['distance'],
0.0,
is_near(results['distance'], 0.0)))
print("{} is near-enough to {}: {}".format(
results['distance'], 0.0, is_near(results['distance'], 0.0)))
results = engines.run(ten_distance)
print(results)
print("%s is near-enough to %s: %s" % (results['distance'],
14.14199,
is_near(results['distance'],
14.14199)))
print("{} is near-enough to {}: {}".format(
results['distance'], 14.14199, is_near(results['distance'], 14.14199)))

4
taskflow/examples/dump_memory_backend.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -67,6 +65,6 @@ print("---------")
for path in backend.memory.ls_r(backend.memory.root_path, absolute=True):
value = backend.memory[path]
if value:
print("%s -> %s" % (path, value))
print("{} -> {}".format(path, value))
else:
print("%s" % (path))

2
taskflow/examples/echo_listener.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/examples/example_utils.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -52,7 +50,7 @@ def rm_path(persist_path):
raise ValueError("Unknown how to `rm` path: %s" % (persist_path))
try:
rm_func(persist_path)
except (IOError, OSError):
except OSError:
pass

20
taskflow/examples/fake_billing.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -47,12 +45,12 @@ from taskflow.utils import misc
# complete to 100% complete.
class DB(object):
class DB:
def query(self, sql):
print("Querying with: %s" % (sql))
class UrlCaller(object):
class UrlCaller:
def __init__(self):
self._send_time = 0.5
self._chunks = 25
@ -73,7 +71,7 @@ class UrlCaller(object):
# 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:
def __init__(self):
self._db_handle = None
self._url_handle = None
@ -93,7 +91,7 @@ class ResourceFetcher(object):
class ExtractInputRequest(task.Task):
def __init__(self, resources):
super(ExtractInputRequest, self).__init__(provides="parsed_request")
super().__init__(provides="parsed_request")
self._resources = resources
def execute(self, request):
@ -106,7 +104,7 @@ class ExtractInputRequest(task.Task):
class MakeDBEntry(task.Task):
def __init__(self, resources):
super(MakeDBEntry, self).__init__()
super().__init__()
self._resources = resources
def execute(self, parsed_request):
@ -120,7 +118,7 @@ class MakeDBEntry(task.Task):
class ActivateDriver(task.Task):
def __init__(self, resources):
super(ActivateDriver, self).__init__(provides='sent_to')
super().__init__(provides='sent_to')
self._resources = resources
self._url = "http://blahblah.com"
@ -138,8 +136,8 @@ class ActivateDriver(task.Task):
def update_progress(self, progress, **kwargs):
# Override the parent method to also print out the status.
super(ActivateDriver, self).update_progress(progress, **kwargs)
print("%s is %0.2f%% done" % (self.name, progress * 100))
super().update_progress(progress, **kwargs)
print("{} is {:0.2f}% done".format(self.name, progress * 100))
class DeclareSuccess(task.Task):
@ -148,7 +146,7 @@ class DeclareSuccess(task.Task):
print("All data processed and sent to %s" % (sent_to))
class DummyUser(object):
class DummyUser:
def __init__(self, user, id_):
self.user = user
self.id = id_

6
taskflow/examples/graph_flow.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -99,7 +97,7 @@ print("Single threaded engine result %s" % result)
for (name, value) in expected:
actual = result.get(name)
if actual != value:
sys.stderr.write("%s != %s\n" % (actual, value))
sys.stderr.write("{} != {}\n".format(actual, value))
unexpected += 1
result = taskflow.engines.run(
@ -109,7 +107,7 @@ print("Multi threaded engine result %s" % result)
for (name, value) in expected:
actual = result.get(name)
if actual != value:
sys.stderr.write("%s != %s\n" % (actual, value))
sys.stderr.write("{} != {}\n".format(actual, value))
unexpected += 1
if unexpected:

6
taskflow/examples/hello_world.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -38,12 +36,12 @@ from taskflow import task
class PrinterTask(task.Task):
def __init__(self, name, show_name=True, inject=None):
super(PrinterTask, self).__init__(name, inject=inject)
super().__init__(name, inject=inject)
self._show_name = show_name
def execute(self, output):
if self._show_name:
print("%s: %s" % (self.name, output))
print("{}: {}".format(self.name, output))
else:
print(output)

8
taskflow/examples/jobboard_produce_consume_colors.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -84,9 +82,9 @@ def dispatch_work(job):
def safe_print(name, message, prefix=""):
with STDOUT_LOCK:
if prefix:
print("%s %s: %s" % (prefix, name, message))
print("{} {}: {}".format(prefix, name, message))
else:
print("%s: %s" % (name, message))
print("{}: {}".format(name, message))
def worker(ident, client, consumed):
@ -138,7 +136,7 @@ def producer(ident, client):
safe_print(name, "started")
with backends.backend(name, SHARED_CONF.copy(), client=client) as board:
for i in range(0, PRODUCER_UNITS):
job_name = "%s-%s" % (name, i)
job_name = "{}-{}".format(name, i)
details = {
'color': random.choice(['red', 'blue']),
}

4
taskflow/examples/parallel_table_multiply.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -47,7 +45,7 @@ class RowMultiplier(task.Task):
"""Performs a modification of an input row, creating a output row."""
def __init__(self, name, index, row, multiplier):
super(RowMultiplier, self).__init__(name=name)
super().__init__(name=name)
self.index = index
self.multiplier = multiplier
self.row = row

4
taskflow/examples/persistence_example.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -58,7 +56,7 @@ class HiTask(task.Task):
class ByeTask(task.Task):
def __init__(self, blowup):
super(ByeTask, self).__init__()
super().__init__()
self._blowup = blowup
def execute(self):

6
taskflow/examples/pseudo_scoping.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Ivan Melnikov <iv at altlinux dot org>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -67,7 +65,7 @@ class CallTask(task.Task):
"""Task that calls person by number."""
def execute(self, person, number):
print('Calling %s %s.' % (person, number))
print('Calling {} {}.'.format(person, number))
# This is how it works for one person:
@ -84,7 +82,7 @@ taskflow.engines.run(simple_flow, store={'person': 'Josh'})
# we use `rebind` argument of task constructor.
def subflow_factory(prefix):
def pr(what):
return '%s-%s' % (prefix, what)
return '{}-{}'.format(prefix, what)
return lf.Flow(pr('flow')).add(
FetchNumberTask(pr('fetch'),

4
taskflow/examples/resume_from_backend.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -62,7 +60,7 @@ import example_utils as eu # noqa
def print_task_states(flowdetail, msg):
eu.print_wrapped(msg)
print("Flow '%s' state: %s" % (flowdetail.name, flowdetail.state))
print("Flow '{}' state: {}".format(flowdetail.name, flowdetail.state))
# Sort by these so that our test validation doesn't get confused by the
# order in which the items in the flow detail can be in.
items = sorted((td.name, td.version, td.state, td.results)

2
taskflow/examples/resume_many_flows.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

2
taskflow/examples/resume_many_flows/my_flows.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/examples/resume_many_flows/resume_all.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -39,7 +37,7 @@ FINISHED_STATES = (states.SUCCESS, states.FAILURE, states.REVERTED)
def resume(flowdetail, backend):
print('Resuming flow %s %s' % (flowdetail.name, flowdetail.uuid))
print('Resuming flow {} {}'.format(flowdetail.name, flowdetail.uuid))
engine = taskflow.engines.load_from_detail(flow_detail=flowdetail,
backend=backend)
engine.run()

6
taskflow/examples/resume_many_flows/run_flow.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -38,6 +36,6 @@ import my_flows # noqa
with example_utils.get_backend() as backend:
engine = taskflow.engines.load_from_factory(my_flows.flow_factory,
backend=backend)
print('Running flow %s %s' % (engine.storage.flow_name,
engine.storage.flow_uuid))
print('Running flow {} {}'.format(engine.storage.flow_name,
engine.storage.flow_uuid))
engine.run()

27
taskflow/examples/resume_vm_boot.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -63,7 +61,7 @@ class PrintText(task.Task):
"""Just inserts some text print outs in a workflow."""
def __init__(self, print_what, no_slow=False):
content_hash = hashlib.md5(print_what.encode('utf-8')).hexdigest()[0:8]
super(PrintText, self).__init__(name="Print: %s" % (content_hash))
super().__init__(name="Print: %s" % (content_hash))
self._text = print_what
self._no_slow = no_slow
@ -78,7 +76,7 @@ class PrintText(task.Task):
class DefineVMSpec(task.Task):
"""Defines a vm specification to be."""
def __init__(self, name):
super(DefineVMSpec, self).__init__(provides='vm_spec', name=name)
super().__init__(provides='vm_spec', name=name)
def execute(self):
return {
@ -93,8 +91,7 @@ class DefineVMSpec(task.Task):
class LocateImages(task.Task):
"""Locates where the vm images are."""
def __init__(self, name):
super(LocateImages, self).__init__(provides='image_locations',
name=name)
super().__init__(provides='image_locations', name=name)
def execute(self, vm_spec):
image_locations = {}
@ -107,13 +104,13 @@ class LocateImages(task.Task):
class DownloadImages(task.Task):
"""Downloads all the vm images."""
def __init__(self, name):
super(DownloadImages, self).__init__(provides='download_paths',
name=name)
super().__init__(provides='download_paths',
name=name)
def execute(self, image_locations):
for src, loc in image_locations.items():
with slow_down(1):
print("Downloading from %s => %s" % (src, loc))
print("Downloading from {} => {}".format(src, loc))
return sorted(image_locations.values())
@ -125,8 +122,8 @@ IPADDR=%s
ONBOOT=yes"""
def __init__(self, name):
super(CreateNetworkTpl, self).__init__(provides='network_settings',
name=name)
super().__init__(provides='network_settings',
name=name)
def execute(self, ips):
settings = []
@ -138,7 +135,7 @@ ONBOOT=yes"""
class AllocateIP(task.Task):
"""Allocates the ips for the given vm."""
def __init__(self, name):
super(AllocateIP, self).__init__(provides='ips', name=name)
super().__init__(provides='ips', name=name)
def execute(self, vm_spec):
ips = []
@ -152,7 +149,7 @@ class WriteNetworkSettings(task.Task):
def execute(self, download_paths, network_settings):
for j, path in enumerate(download_paths):
with slow_down(1):
print("Mounting %s to /tmp/%s" % (path, j))
print("Mounting {} to /tmp/{}".format(path, j))
for i, setting in enumerate(network_settings):
filename = ("/tmp/etc/sysconfig/network-scripts/"
"ifcfg-eth%s" % (i))
@ -263,8 +260,8 @@ with eu.get_backend() as backend:
backend=backend, book=book,
engine='parallel',
executor=executor)
print("!! Your tracking id is: '%s+%s'" % (book.uuid,
engine.storage.flow_uuid))
print("!! Your tracking id is: '{}+{}'".format(
book.uuid, engine.storage.flow_uuid))
print("!! Please submit this on later runs for tracking purposes")
else:
# Attempt to load from a previously partially completed flow.

8
taskflow/examples/resume_volume_create.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -72,7 +70,7 @@ def find_flow_detail(backend, book_id, flow_id):
class PrintText(task.Task):
def __init__(self, print_what, no_slow=False):
content_hash = hashlib.md5(print_what.encode('utf-8')).hexdigest()[0:8]
super(PrintText, self).__init__(name="Print: %s" % (content_hash))
super().__init__(name="Print: %s" % (content_hash))
self._text = print_what
self._no_slow = no_slow
@ -141,8 +139,8 @@ with example_utils.get_backend() as backend:
book.add(flow_detail)
with contextlib.closing(backend.get_connection()) as conn:
conn.save_logbook(book)
print("!! Your tracking id is: '%s+%s'" % (book.uuid,
flow_detail.uuid))
print("!! Your tracking id is: '{}+{}'".format(book.uuid,
flow_detail.uuid))
print("!! Please submit this on later runs for tracking purposes")
else:
flow_detail = find_flow_detail(backend, book_id, flow_id)

2
taskflow/examples/retry_flow.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/examples/reverting_linear.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -59,7 +57,7 @@ class CallJoe(task.Task):
class CallSuzzie(task.Task):
def execute(self, suzzie_number, *args, **kwargs):
raise IOError("Suzzie not home right now.")
raise OSError("Suzzie not home right now.")
# Create your flow and associated tasks (the work to be done).

2
taskflow/examples/run_by_iter.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/examples/run_by_iter_enumerate.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -51,4 +49,4 @@ e.compile()
e.prepare()
for i, st in enumerate(e.run_iter(), 1):
print("Transition %s: %s" % (i, st))
print("Transition {}: {}".format(i, st))

6
taskflow/examples/share_engine_thread.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -43,11 +41,11 @@ from taskflow.utils import threading_utils as tu
class DelayedTask(task.Task):
def __init__(self, name):
super(DelayedTask, self).__init__(name=name)
super().__init__(name=name)
self._wait_for = random.random()
def execute(self):
print("Running '%s' in thread '%s'" % (self.name, tu.get_ident()))
print("Running '{}' in thread '{}'".format(self.name, tu.get_ident()))
time.sleep(self._wait_for)

2
taskflow/examples/simple_linear.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/examples/simple_linear_listening.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -71,7 +69,7 @@ def flow_watch(state, details):
def task_watch(state, details):
print('Task %s => %s' % (details.get('task_name'), state))
print('Task {} => {}'.format(details.get('task_name'), state))
# Wrap your functions into a task type that knows how to treat your functions

2
taskflow/examples/simple_linear_pass.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

2
taskflow/examples/simple_map_reduce.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

4
taskflow/examples/switch_graph_flow.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -64,7 +62,7 @@ while entries:
path = entries.pop()
value = backend.memory[path]
if value:
print("%s -> %s" % (path, value))
print("{} -> {}".format(path, value))
else:
print("%s" % (path))
entries.extend(os.path.join(path, child)

4
taskflow/examples/timing_listener.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -45,7 +43,7 @@ from taskflow import task
class VariableTask(task.Task):
def __init__(self, name):
super(VariableTask, self).__init__(name)
super().__init__(name)
self._sleepy_time = random.random()
def execute(self):

8
taskflow/examples/tox_conductor.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -60,7 +58,7 @@ from taskflow.utils import threading_utils
RUN_TIME = 5
REVIEW_CREATION_DELAY = 0.5
SCAN_DELAY = 0.1
NAME = "%s_%s" % (socket.getfqdn(), os.getpid())
NAME = "{}_{}".format(socket.getfqdn(), os.getpid())
# This won't really use zookeeper but will use a local version of it using
# the zake library that mimics an actual zookeeper cluster using threads and
@ -74,7 +72,7 @@ class RunReview(task.Task):
# A dummy task that clones the review and runs tox...
def _clone_review(self, review, temp_dir):
print("Cloning review '%s' into %s" % (review['id'], temp_dir))
print("Cloning review '{}' into {}".format(review['id'], temp_dir))
def _run_tox(self, temp_dir):
print("Running tox in %s" % temp_dir)
@ -177,7 +175,7 @@ def generate_reviewer(client, saver, name=NAME):
'review': review,
},
}
job_name = "%s_%s" % (real_name, review['id'])
job_name = "{}_{}".format(real_name, review['id'])
print("Posting review '%s'" % review['id'])
jb.post(job_name,
book=make_save_book(saver, review['id']),

2
taskflow/examples/wbe_event_sender.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

2
taskflow/examples/wbe_mandelbrot.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

2
taskflow/examples/wbe_simple_linear.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

2
taskflow/examples/wrapped_exception.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

16
taskflow/exceptions.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -67,7 +65,7 @@ class TaskFlowException(Exception):
this is not yet implemented/supported natively.
"""
def __init__(self, message, cause=None):
super(TaskFlowException, self).__init__(message)
super().__init__(message)
self._cause = cause
@property
@ -192,7 +190,7 @@ class MissingDependencies(DependencyFailure):
message = self.MESSAGE_TPL % {'who': who, 'requirements': requirements}
if method:
message = (self.METHOD_TPL % {'method': method}) + message
super(MissingDependencies, self).__init__(message, cause=cause)
super().__init__(message, cause=cause)
self.missing_requirements = requirements
@ -228,7 +226,7 @@ class DisallowedAccess(TaskFlowException):
"""Raised when storage access is not possible due to state limitations."""
def __init__(self, message, cause=None, state=None):
super(DisallowedAccess, self).__init__(message, cause=cause)
super().__init__(message, cause=cause)
self.state = state
@ -261,7 +259,7 @@ class WrappedFailure(Exception):
"""
def __init__(self, causes):
super(WrappedFailure, self).__init__()
super().__init__()
self._causes = []
for cause in causes:
if cause.check(type(self)) and cause.exception:
@ -306,8 +304,8 @@ class WrappedFailure(Exception):
def __str__(self):
buf = io.StringIO()
buf.write(u'WrappedFailure: [')
buf.write('WrappedFailure: [')
causes_gen = (str(cause) for cause in self._causes)
buf.write(u", ".join(causes_gen))
buf.write(u']')
buf.write(", ".join(causes_gen))
buf.write(']')
return buf.getvalue()

4
taskflow/flow.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -42,7 +40,7 @@ _CHOP_PAT_LEN = len(_CHOP_PAT)
LINK_DECIDER_DEPTH = 'decider_depth'
class Flow(object, metaclass=abc.ABCMeta):
class Flow(metaclass=abc.ABCMeta):
"""The base abstract class of all flow implementations.
A flow is a structure that defines relationships between tasks. You can

9
taskflow/formatters.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -58,7 +56,7 @@ def _fetch_predecessor_tree(graph, atom):
return root
class FailureFormatter(object):
class FailureFormatter:
"""Formats a failure and connects it to associated atoms & engine."""
_BUILDERS = {
@ -107,7 +105,7 @@ class FailureFormatter(object):
if provides_found:
atom_attrs['provides'] = provides
if atom_attrs:
return "Atom '%s' %s" % (atom_name, atom_attrs)
return "Atom '{}' {}".format(atom_name, atom_attrs)
else:
return "Atom '%s'" % (atom_name)
else:
@ -156,7 +154,8 @@ class FailureFormatter(object):
builder, kind = self._BUILDERS[atom_intention]
rooted_tree = builder(graph, atom)
child_count = rooted_tree.child_count(only_direct=False)
buff.write_nl('%s %s (most recent first):' % (child_count, kind))
buff.write_nl(
'{} {} (most recent first):'.format(child_count, kind))
formatter = functools.partial(self._format_node, storage, cache)
direct_child_count = rooted_tree.child_count(only_direct=True)
for i, child in enumerate(rooted_tree, 1):

2
taskflow/jobs/backends/__init__.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

12
taskflow/jobs/backends/impl_redis.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -69,10 +67,10 @@ class RedisJob(base.Job):
created_on=None, backend=None,
book=None, book_data=None,
priority=base.JobPriority.NORMAL):
super(RedisJob, self).__init__(board, name,
uuid=uuid, details=details,
backend=backend,
book=book, book_data=book_data)
super().__init__(board, name,
uuid=uuid, details=details,
backend=backend,
book=book, book_data=book_data)
self._created_on = created_on
self._client = board._client
self._redis_version = board._redis_version
@ -599,7 +597,7 @@ return cmsgpack.pack(result)
def __init__(self, name, conf,
client=None, persistence=None):
super(RedisJobBoard, self).__init__(name, conf)
super().__init__(name, conf)
self._closed = True
if client is not None:
self._client = client

15
taskflow/jobs/backends/impl_zookeeper.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -51,10 +49,10 @@ class ZookeeperJob(base.Job):
uuid=None, details=None, book=None, book_data=None,
created_on=None, backend=None,
priority=base.JobPriority.NORMAL):
super(ZookeeperJob, self).__init__(board, name,
uuid=uuid, details=details,
backend=backend,
book=book, book_data=book_data)
super().__init__(board, name,
uuid=uuid, details=details,
backend=backend,
book=book, book_data=book_data)
self._client = client
self._path = k_paths.normpath(path)
self._lock_path = self._path + board.LOCK_POSTFIX
@ -281,7 +279,7 @@ class ZookeeperJobBoard(base.NotifyingJobBoard):
def __init__(self, name, conf,
client=None, persistence=None, emit_notifications=True):
super(ZookeeperJobBoard, self).__init__(name, conf)
super().__init__(name, conf)
if client is not None:
self._client = client
self._owned = False
@ -552,7 +550,8 @@ class ZookeeperJobBoard(base.NotifyingJobBoard):
except Exception:
owner = None
if owner:
message = "Job %s already claimed by '%s'" % (job.uuid, owner)
message = "Job {} already claimed by '{}'".format(
job.uuid, owner)
else:
message = "Job %s already claimed" % (job.uuid)
excp.raise_with_cause(excp.UnclaimableJob,

12
taskflow/jobs/base.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Rackspace Hosting Inc. All Rights Reserved.
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
@ -110,7 +108,7 @@ class JobPriority(enum.Enum):
return tuple(values)
class Job(object, metaclass=abc.ABCMeta):
class Job(metaclass=abc.ABCMeta):
"""A abstraction that represents a named and trackable unit of work.
A job connects a logbook, a owner, a priority, last modified and created
@ -277,12 +275,12 @@ class Job(object, metaclass=abc.ABCMeta):
def __str__(self):
"""Pretty formats the job into something *more* meaningful."""
cls_name = type(self).__name__
return "%s: %s (priority=%s, uuid=%s, details=%s)" % (
return "{}: {} (priority={}, uuid={}, details={})".format(
cls_name, self.name, self.priority,
self.uuid, self.details)
class JobBoardIterator(object):
class JobBoardIterator:
"""Iterator over a jobboard that iterates over potential jobs.
It provides the following attributes:
@ -355,7 +353,7 @@ class JobBoardIterator(object):
return job
class JobBoard(object, metaclass=abc.ABCMeta):
class JobBoard(metaclass=abc.ABCMeta):
"""A place where jobs can be posted, reposted, claimed and transferred.
There can be multiple implementations of this job board, depending on the
@ -565,7 +563,7 @@ class NotifyingJobBoard(JobBoard):
registered are thread safe (and block for as little time as possible).
"""
def __init__(self, name, conf):
super(NotifyingJobBoard, self).__init__(name, conf)
super().__init__(name, conf)
self.notifier = notifier.Notifier()

4
taskflow/listeners/base.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -77,7 +75,7 @@ def _bulk_register(watch_states, notifier, cb, details_filter=None):
return registered
class Listener(object):
class Listener:
"""Base class for listeners.
A listener can be attached to an engine to do various actions on flow and

4
taskflow/listeners/capturing.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -65,7 +63,7 @@ class CaptureListener(base.Listener):
# Provide your own list (or previous list) to accumulate
# into...
values=None):
super(CaptureListener, self).__init__(
super().__init__(
engine,
task_listen_for=task_listen_for,
flow_listen_for=flow_listen_for,

4
taskflow/listeners/claims.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -49,7 +47,7 @@ class CheckingClaimListener(base.Listener):
"""
def __init__(self, engine, job, board, owner, on_job_loss=None):
super(CheckingClaimListener, self).__init__(engine)
super().__init__(engine)
self._job = job
self._board = board
self._owner = owner

6
taskflow/listeners/logging.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -46,7 +44,7 @@ class LoggingListener(base.DumpingListener):
retry_listen_for=base.DEFAULT_LISTEN_FOR,
log=None,
level=logging.DEBUG):
super(LoggingListener, self).__init__(
super().__init__(
engine, task_listen_for=task_listen_for,
flow_listen_for=flow_listen_for, retry_listen_for=retry_listen_for)
self._logger = misc.pick_first_not_none(log, self._LOGGER, LOG)
@ -111,7 +109,7 @@ class DynamicLoggingListener(base.Listener):
log=None, failure_level=logging.WARNING,
level=logging.DEBUG, hide_inputs_outputs_of=(),
fail_formatter=None):
super(DynamicLoggingListener, self).__init__(
super().__init__(
engine, task_listen_for=task_listen_for,
flow_listen_for=flow_listen_for, retry_listen_for=retry_listen_for)
self._failure_level = failure_level

4
taskflow/listeners/printing.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -27,7 +25,7 @@ class PrintingListener(base.DumpingListener):
flow_listen_for=base.DEFAULT_LISTEN_FOR,
retry_listen_for=base.DEFAULT_LISTEN_FOR,
stderr=False):
super(PrintingListener, self).__init__(
super().__init__(
engine, task_listen_for=task_listen_for,
flow_listen_for=flow_listen_for, retry_listen_for=retry_listen_for)
if stderr:

23
taskflow/listeners/timing.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -26,7 +24,7 @@ from taskflow import logging
from taskflow import states
STARTING_STATES = frozenset((states.RUNNING, states.REVERTING))
FINISHED_STATES = frozenset((base.FINISH_STATES + (states.REVERTED,)))
FINISHED_STATES = frozenset(base.FINISH_STATES + (states.REVERTED,))
WATCH_STATES = frozenset(itertools.chain(FINISHED_STATES, STARTING_STATES,
[states.PENDING]))
@ -48,13 +46,13 @@ class DurationListener(base.Listener):
to task metadata with key ``'duration'``.
"""
def __init__(self, engine):
super(DurationListener, self).__init__(engine,
task_listen_for=WATCH_STATES,
flow_listen_for=WATCH_STATES)
super().__init__(engine,
task_listen_for=WATCH_STATES,
flow_listen_for=WATCH_STATES)
self._timers = {co.TASK: {}, co.FLOW: {}}
def deregister(self):
super(DurationListener, self).deregister()
super().deregister()
# There should be none that still exist at deregistering time, so log a
# warning if there were any that somehow still got left behind...
for item_type, timers in self._timers.items():
@ -105,23 +103,22 @@ class PrintingDurationListener(DurationListener):
"""Listener that prints the duration as well as recording it."""
def __init__(self, engine, printer=None):
super(PrintingDurationListener, self).__init__(engine)
super().__init__(engine)
if printer is None:
self._printer = _printer
else:
self._printer = printer
def _record_ending(self, timer, item_type, item_name, state):
super(PrintingDurationListener, self)._record_ending(
super()._record_ending(
timer, item_type, item_name, state)
self._printer("It took %s '%s' %0.2f seconds to"
" finish." % (item_type, item_name, timer.elapsed()))
def _receiver(self, item_type, item_name, state):
super(PrintingDurationListener, self)._receiver(item_type,
item_name, state)
super()._receiver(item_type, item_name, state)
if state in STARTING_STATES:
self._printer("'%s' %s started." % (item_name, item_type))
self._printer("'{}' {} started.".format(item_name, item_type))
class EventTimeListener(base.Listener):
@ -139,7 +136,7 @@ class EventTimeListener(base.Listener):
task_listen_for=base.DEFAULT_LISTEN_FOR,
flow_listen_for=base.DEFAULT_LISTEN_FOR,
retry_listen_for=base.DEFAULT_LISTEN_FOR):
super(EventTimeListener, self).__init__(
super().__init__(
engine, task_listen_for=task_listen_for,
flow_listen_for=flow_listen_for, retry_listen_for=retry_listen_for)

2
taskflow/logging.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

6
taskflow/patterns/graph_flow.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -66,7 +64,7 @@ class Flow(flow.Flow):
"""
def __init__(self, name, retry=None):
super(Flow, self).__init__(name, retry)
super().__init__(name, retry)
self._graph = gr.DiGraph(name=name)
self._graph.freeze()
@ -332,7 +330,7 @@ class TargetedFlow(Flow):
"""
def __init__(self, *args, **kwargs):
super(TargetedFlow, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
self._subgraph = None
self._target = None

13
taskflow/patterns/linear_flow.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -36,7 +34,7 @@ class Flow(flow.Flow):
"""
def __init__(self, name, retry=None):
super(Flow, self).__init__(name, retry)
super().__init__(name, retry)
self._graph = gr.OrderedDiGraph(name=name)
self._last_item = self._no_last_item
@ -55,8 +53,7 @@ class Flow(flow.Flow):
return len(self._graph)
def __iter__(self):
for item in self._graph.nodes:
yield item
yield from self._graph.nodes
@property
def requires(self):
@ -71,9 +68,7 @@ class Flow(flow.Flow):
return frozenset(requires)
def iter_nodes(self):
for (n, n_data) in self._graph.nodes(data=True):
yield (n, n_data)
yield from self._graph.nodes(data=True)
def iter_links(self):
for (u, v, e_data) in self._graph.edges(data=True):
yield (u, v, e_data)
yield from self._graph.edges(data=True)

13
taskflow/patterns/unordered_flow.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -26,7 +24,7 @@ class Flow(flow.Flow):
"""
def __init__(self, name, retry=None):
super(Flow, self).__init__(name, retry)
super().__init__(name, retry)
self._graph = gr.Graph(name=name)
def add(self, *items):
@ -40,16 +38,13 @@ class Flow(flow.Flow):
return len(self._graph)
def __iter__(self):
for item in self._graph:
yield item
yield from self._graph
def iter_links(self):
for (u, v, e_data) in self._graph.edges(data=True):
yield (u, v, e_data)
yield from self._graph.edges(data=True)
def iter_nodes(self):
for n, n_data in self._graph.nodes(data=True):
yield (n, n_data)
yield from self._graph.nodes(data=True)
@property
def requires(self):

4
taskflow/persistence/backends/__init__.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2013 Rackspace Hosting Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -64,7 +62,7 @@ def fetch(conf, namespace=BACKEND_NAMESPACE, **kwargs):
invoke_kwds=kwargs)
return mgr.driver
except RuntimeError as e:
raise exc.NotFound("Could not find backend %s: %s" % (backend, e))
raise exc.NotFound("Could not find backend {}: {}".format(backend, e))
@contextlib.contextmanager

9
taskflow/persistence/backends/impl_dir.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
# Copyright (C) 2013 Rackspace Hosting All Rights Reserved.
#
@ -17,7 +15,6 @@
import contextlib
import errno
import io
import os
import shutil
@ -69,7 +66,7 @@ class DirBackend(path_based.PathBasedBackend):
"""
def __init__(self, conf):
super(DirBackend, self).__init__(conf)
super().__init__(conf)
max_cache_size = self._conf.get('max_cache_size')
if max_cache_size is not None:
max_cache_size = int(max_cache_size)
@ -100,7 +97,7 @@ class Connection(path_based.PathBasedConnection):
mtime = os.path.getmtime(filename)
cache_info = self.backend.file_cache.setdefault(filename, {})
if not cache_info or mtime > cache_info.get('mtime', 0):
with io.open(filename, 'r', encoding=self.backend.encoding) as fp:
with open(filename, encoding=self.backend.encoding) as fp:
cache_info['data'] = fp.read()
cache_info['mtime'] = mtime
return cache_info['data']
@ -108,7 +105,7 @@ class Connection(path_based.PathBasedConnection):
def _write_to(self, filename, contents):
contents = misc.binary_encode(contents,
encoding=self.backend.encoding)
with io.open(filename, 'wb') as fp:
with open(filename, 'wb') as fp:
fp.write(contents)
self.backend.file_cache.pop(filename, None)

15
taskflow/persistence/backends/impl_memory.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
# Copyright (C) 2013 Rackspace Hosting All Rights Reserved.
#
@ -31,10 +29,10 @@ class FakeInode(tree.Node):
"""A in-memory filesystem inode-like object."""
def __init__(self, item, path, value=None):
super(FakeInode, self).__init__(item, path=path, value=value)
super().__init__(item, path=path, value=value)
class FakeFilesystem(object):
class FakeFilesystem:
"""An in-memory filesystem-like structure.
This filesystem uses posix style paths **only** so users must be careful
@ -249,8 +247,7 @@ class FakeFilesystem(object):
parts = path.split(pp.sep)[1:]
if include_root:
parts.insert(0, self._root.item)
for piece in parts:
yield piece
yield from parts
def __delitem__(self, path):
self.delete(path, recursive=True)
@ -258,7 +255,7 @@ class FakeFilesystem(object):
@staticmethod
def _stringify_node(node):
if 'target' in node.metadata:
return "%s (link to %s)" % (node.item, node.metadata['target'])
return "{} (link to {})".format(node.item, node.metadata['target'])
else:
return str(node.item)
@ -309,7 +306,7 @@ class MemoryBackend(path_based.PathBasedBackend):
DEFAULT_PATH = pp.sep
def __init__(self, conf=None):
super(MemoryBackend, self).__init__(conf)
super().__init__(conf)
self.memory = FakeFilesystem(deep_copy=self._conf.get('deep_copy',
True))
self.lock = fasteners.ReaderWriterLock()
@ -323,7 +320,7 @@ class MemoryBackend(path_based.PathBasedBackend):
class Connection(path_based.PathBasedConnection):
def __init__(self, backend):
super(Connection, self).__init__(backend)
super().__init__(backend)
self.upgrade()
@contextlib.contextmanager

10
taskflow/persistence/backends/impl_sqlalchemy.py
View File

@ -179,7 +179,7 @@ def _ping_listener(dbapi_conn, connection_rec, connection_proxy):
raise
class _Alchemist(object):
class _Alchemist:
"""Internal <-> external row <-> objects + other helper functions.
NOTE(harlowja): for internal usage only.
@ -235,7 +235,7 @@ class SQLAlchemyBackend(base.Backend):
}
"""
def __init__(self, conf, engine=None):
super(SQLAlchemyBackend, self).__init__(conf)
super().__init__(conf)
if engine is not None:
self._engine = engine
self._owns_engine = False
@ -581,8 +581,7 @@ class Connection(base.Connection):
exc.raise_with_cause(exc.StorageFailure,
"Failed getting flow details in"
" logbook '%s'" % book_uuid)
for flow_details in gathered:
yield flow_details
yield from gathered
def get_flow_details(self, fd_uuid, lazy=False):
try:
@ -631,8 +630,7 @@ class Connection(base.Connection):
exc.raise_with_cause(exc.StorageFailure,
"Failed getting atom details in flow"
" detail '%s'" % fd_uuid)
for atom_details in gathered:
yield atom_details
yield from gathered
def close(self):
pass

6
taskflow/persistence/backends/impl_zookeeper.py
View File

@ -1,5 +1,3 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 AT&T Labs All Rights Reserved.
# Copyright (C) 2015 Rackspace Hosting All Rights Reserved.
#
@ -56,7 +54,7 @@ class ZkBackend(path_based.PathBasedBackend):
DEFAULT_PATH = '/taskflow'
def __init__(self, conf, client=None):
super(ZkBackend, self).__init__(conf)
super().__init__(conf)
if not paths.isabs(self._path):
raise ValueError("Zookeeper path must be absolute")
if client is not None:
@ -87,7 +85,7 @@ class ZkBackend(path_based.PathBasedBackend):
class ZkConnection(path_based.PathBasedConnection):
def __init__(self, backend, client, conf):
super(ZkConnection, self).__init__(backend)
super().__init__(backend)
self._conf = conf
self._client = client
with self._exc_wrapper():

Some files were not shown because too many files have changed in this diff Show More