https://prefect.io logo
Powered by
Title
w

Wei Mei

06/29/2022, 9:21 PM
Hi, having a complete brainfart moment. I am trying to return the string “full”, to the next task, but I am getting None: 
@task(nout=1)
def determine_run(clock):
    if clock == "full_run":
        <http://logger.info|logger.info>("full_run clock")
        run_type = "full"
        <http://logger.info|logger.info>(f"{run_type}")
    return run_type
✅ 1
j

Jessica Smith

06/29/2022, 9:44 PM
Is checkpointing enabled?
Also, I don't think you need nout if you're only returning one value
✅ 1
w

Wei Mei

06/29/2022, 9:49 PM
dbt_run = dbt(
_command_=f"dbt run -s {run_type.result}",
_task_args_={"name": f"DBT Run: {run_type.result}"},
_upstream_tasks_=[deps_output])
this is where I am trying to get the string.
right, the nout was a desperate move 😳
not sure about checkpointing, default configuration
j

Jessica Smith

06/29/2022, 9:53 PM
Yeah, been there with the desperate moves. If that code snippet is the init method, you can just pass checkpoint = True to it, that should fix it
Checkpointing is disabled by default, which drove me crazy for a long while, couldn't figure out why nothing was working. Not sure why it isn't enabled by default
You can also set it in your config.toml in your prefect settings, or in an environment variable, and that will enable it globally
w

Wei Mei

06/29/2022, 9:55 PM
@task(name="Determine run type", checkpoint=True)
def determine_run(clock):
    if clock == "full_run":
        <http://logger.info|logger.info>("full_run clock")
        run_type = "small_exchange"
        <http://logger.info|logger.info>(f"{run_type}")

    return run_type
like this?
i added another task to print the result: 
@task(name="Determine run type", checkpoint=True)
def determine_run(clock):
    if clock == "full_run":
        <http://logger.info|logger.info>("full_run clock")
        run_type = "full"
        <http://logger.info|logger.info>(f"{run_type}")

    return run_type

@task(name="Print run result")
def print_result(x):
    <http://logger.info|logger.info>(x)

as flow:
clock = Parameter(name="clock", default="full_run")
run_type = determine_run(clock=clock)
print = print_result(run_type)

dbt_run = dbt(
    command=f"dbt run -s {run_type.result}",
    task_args={"name": f"DBT Run: {run_type.result}"},
    upstream_tasks=[deps_output])
the print_result prints out what I am expecting.
j

Jessica Smith

06/29/2022, 9:57 PM
Yeah, that should work
w

Wei Mei

06/29/2022, 9:58 PM
sorry, should have started with the entire code.
j

Jessica Smith

06/29/2022, 9:58 PM
Does it work when you do it the way you had it setup originally, if you add the checkpoint = True?
w

Wei Mei

06/29/2022, 9:58 PM
it doesnt 😞
print result i get: full
but in dbt_run, I get None
j

Jessica Smith

06/29/2022, 10:01 PM
In print result you are just passing the run_type variable, but in the dbt_run you are passing run_type.result. What if you change it to just run_type?
w

Wei Mei

06/29/2022, 10:02 PM
with just run_type I get 
<Task: Determine run type>
j

Jessica Smith

06/29/2022, 10:03 PM
Hmm... I wonder if this is happening because you are using it in a format string, instead of passing it directly
You might try wrapping the dbt task in your own task method, and have that method receive the run_type argument. Then pass it along to the dbt task by calling the run method
Not at a computer so I can't provide an example, but hopefully that made sense
w

Wei Mei

06/29/2022, 10:05 PM
gotcha.
i did that at first, was trying to make it cleaner using the dbtShellTask
thank you so much for your time.
👍 1
j

Jessica Smith

06/29/2022, 10:07 PM
Sure thing! And i bet that would have worked if checkpointing was enabled by default.
:upvote: 1
6 Views
#prefect-communityJoin Slack
Powered by