1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
|
===========================
Notifications and listeners
===========================
.. testsetup::
from taskflow import task
from taskflow.patterns import linear_flow
from taskflow import engines
from taskflow.types import notifier
ANY = notifier.Notifier.ANY
--------
Overview
--------
Engines provide a way to receive notification on task and flow state
transitions, which is useful for monitoring, logging, metrics, debugging
and plenty of other tasks.
To receive these notifications you should register a callback with
an instance of the :py:class:`~taskflow.types.notifier.Notifier`
class that is attached to :py:class:`~taskflow.engines.base.Engine`
attributes ``atom_notifier`` and ``notifier``.
TaskFlow also comes with a set of predefined :ref:`listeners <listeners>`, and
provides means to write your own listeners, which can be more convenient than
using raw callbacks.
--------------------------------------
Receiving notifications with callbacks
--------------------------------------
Flow notifications
------------------
To receive notification on flow state changes use the
:py:class:`~taskflow.types.notifier.Notifier` instance available as the
``notifier`` property of an engine.
A basic example is:
.. doctest::
>>> class CatTalk(task.Task):
... def execute(self, meow):
... print(meow)
... return "cat"
...
>>> class DogTalk(task.Task):
... def execute(self, woof):
... print(woof)
... return 'dog'
...
>>> def flow_transition(state, details):
... print("Flow '%s' transition to state %s" % (details['flow_name'], state))
...
>>>
>>> flo = linear_flow.Flow("cat-dog").add(
... CatTalk(), DogTalk(provides="dog"))
>>> eng = engines.load(flo, store={'meow': 'meow', 'woof': 'woof'})
>>> eng.notifier.register(ANY, flow_transition)
>>> eng.run()
Flow 'cat-dog' transition to state RUNNING
meow
woof
Flow 'cat-dog' transition to state SUCCESS
Task notifications
------------------
To receive notification on task state changes use the
:py:class:`~taskflow.types.notifier.Notifier` instance available as the
``atom_notifier`` property of an engine.
A basic example is:
.. doctest::
>>> class CatTalk(task.Task):
... def execute(self, meow):
... print(meow)
... return "cat"
...
>>> class DogTalk(task.Task):
... def execute(self, woof):
... print(woof)
... return 'dog'
...
>>> def task_transition(state, details):
... print("Task '%s' transition to state %s" % (details['task_name'], state))
...
>>>
>>> flo = linear_flow.Flow("cat-dog")
>>> flo.add(CatTalk(), DogTalk(provides="dog"))
<taskflow.patterns.linear_flow.Flow object at 0x...>
>>> eng = engines.load(flo, store={'meow': 'meow', 'woof': 'woof'})
>>> eng.task_notifier.register(ANY, task_transition)
>>> eng.run()
Task 'CatTalk' transition to state RUNNING
meow
Task 'CatTalk' transition to state SUCCESS
Task 'DogTalk' transition to state RUNNING
woof
Task 'DogTalk' transition to state SUCCESS
.. _listeners:
---------
Listeners
---------
TaskFlow comes with a set of predefined listeners -- helper classes that can be
used to do various actions on flow and/or tasks transitions. You can also
create your own listeners easily, which may be more convenient than using raw
callbacks for some use cases.
For example, this is how you can use
:py:class:`~taskflow.listeners.printing.PrintingListener`:
.. doctest::
>>> from taskflow.listeners import printing
>>> class CatTalk(task.Task):
... def execute(self, meow):
... print(meow)
... return "cat"
...
>>> class DogTalk(task.Task):
... def execute(self, woof):
... print(woof)
... return 'dog'
...
>>>
>>> flo = linear_flow.Flow("cat-dog").add(
... CatTalk(), DogTalk(provides="dog"))
>>> eng = engines.load(flo, store={'meow': 'meow', 'woof': 'woof'})
>>> with printing.PrintingListener(eng):
... eng.run()
...
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved flow 'cat-dog' (...) into state 'RUNNING' from state 'PENDING'
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved task 'CatTalk' (...) into state 'RUNNING' from state 'PENDING'
meow
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved task 'CatTalk' (...) into state 'SUCCESS' from state 'RUNNING' with result 'cat' (failure=False)
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved task 'DogTalk' (...) into state 'RUNNING' from state 'PENDING'
woof
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved task 'DogTalk' (...) into state 'SUCCESS' from state 'RUNNING' with result 'dog' (failure=False)
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved flow 'cat-dog' (...) into state 'SUCCESS' from state 'RUNNING'
Basic listener
--------------
.. autoclass:: taskflow.listeners.base.Listener
Printing and logging listeners
------------------------------
.. autoclass:: taskflow.listeners.base.DumpingListener
.. autoclass:: taskflow.listeners.logging.LoggingListener
.. autoclass:: taskflow.listeners.logging.DynamicLoggingListener
.. autoclass:: taskflow.listeners.printing.PrintingListener
Timing listener
---------------
.. autoclass:: taskflow.listeners.timing.TimingListener
.. autoclass:: taskflow.listeners.timing.PrintingTimingListener
Claim listener
--------------
.. autoclass:: taskflow.listeners.claims.CheckingClaimListener
Hierarchy
---------
.. inheritance-diagram::
taskflow.listeners.base.DumpingListener
taskflow.listeners.base.Listener
taskflow.listeners.claims.CheckingClaimListener
taskflow.listeners.logging.DynamicLoggingListener
taskflow.listeners.logging.LoggingListener
taskflow.listeners.printing.PrintingListener
taskflow.listeners.timing.PrintingTimingListener
taskflow.listeners.timing.TimingListener
:parts: 1
|
