I have a question about the task decorator

@task
def get_file_name(table):
    file_name = f'bpi/{table}/-000'
    return file_name

with Flow("write to s3") as flow:
    exports = ['email_acquisitions', 'new_list_acquisitions', 'revenue', 'segmentation']
    for export in exports:
        file_name = get_file_name(export)
        print(file_name)

I was trying to write a function that does some string manipulation to get a s3 path (this is simplified ^) and when I add the @task decorator,

file_name

= <Task: get_file_name>, when I just want it to return the string produced. Am i doing something in a non-prefect way?

Hi @Jacob (he/him) - your code should be doing what you hope; the difference is that your task calls are not evaluated at script runtime but instead are used to create the metadata for your flow. Since

print

is evaluated at script runtime, it's outputting the task method instead of the task evaluation. If you modify your code like this:

@task
def get_file_name(table):
    file_name = f'bpi/{table}/-000'
    return file_name

@task
def print_file_name(file):
    print(file)

with Flow("write to s3") as flow:
    exports = ['email_acquisitions', 'new_list_acquisitions', 'revenue', 'segmentation']
    for export in exports:
        file_name = get_file_name(export)
        print_file_name(file_name)

flow.run()

you'll be able to see what you're expecting

My goal isn’t actually to print the file, i was just using it to debug, but good rec on loggers. The problem I’m having is trying to use the string downstream and still receiving it as a task object.

oh! i'm sorry, i completely ignored the rest of your code. you'll want to map over that list instead of using a for loop in the flow context:

@task
def get_file_name(table):
    file_name = f'bpi/{table}/-000'
    return file_name

@task
def print_file_name(file):
    print(file)

@task export_list():
    return ['email_acquisitions', 'new_list_acquisitions', 'revenue', 'segmentation']

with Flow("write to s3") as flow:
    file_name = get_file_name.map(export_list)
    print_file_name.map(file_name)
    
flow.run()

oh interesting, thanks!