Title
c

Carlos Paiva

12/29/2021, 5:41 PM
Hi All, Another noob question: having the given Flow
with Flow(name="name", storage=storage, run_config=run_config) as flow:
    # Pipeline parameters
    event = Parameter('event', required=True, default={})

    x = event.get("body", None)
How can I access the dictionary data? I am getting
AttributeError: 'Parameter' object has no attribute 'get'
n

nicholas

12/29/2021, 5:51 PM
Hi @Carlos Paiva -
event
in your flow context is a reference to the Parameter task; the dictionary won’t be available until runtime, when
event
is passed to another task. To use the
event
parameter, you can pass it to another task like this:
@task
def some_task(event):
  x = event.get("body", None)

with Flow(name="name") as flow:
    # Pipeline parameters
    event = Parameter('event', required=True, default={})
    some_task(event)
c

Carlos Paiva

12/30/2021, 2:18 PM
Thank you @nicholas! This works! However, how can we control Task execution from within the
with Flow
? Saying that we return
x
directly from
some_task
it will get
FunctionTask
type. We’re not able to do
if x.get("key") == "value":
another_task()
n

nicholas

12/30/2021, 5:08 PM
This might help! Basically you can use conditional logic by using a
case
context block, wrapping the reference to your parameter/task
c

Carlos Paiva

01/03/2022, 6:49 PM
Thank you, @nicholas !
👍 1