https://prefect.io
Join Slack
Hi, having a complete brainfart moment. I am trying to return the string “full”, to the next task, b...
# ask-community
w
Hi, having a complete brainfart moment. I am trying to return the string “full”, to the next task, but I am getting None:
Copy code
@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
Is checkpointing enabled?
Also, I don't think you need nout if you're only returning one value
✅ 1
w
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
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
Copy code
@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:
Copy code
@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
Yeah, that should work
w
sorry, should have started with the entire code.
j
Does it work when you do it the way you had it setup originally, if you add the checkpoint = True?
w
it doesnt 😞
print result i get: full
but in dbt_run, I get None
j
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
with just run_type I get 
<Task: Determine run type>
j
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
gotcha.
i did that at first, was trying to make it cleaner using the dbtShellTask
thank you so much for your time.
👍 1
j
Sure thing! And i bet that would have worked if checkpointing was enabled by default.
upvote 1
5 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by