I have a question about the task decorator ```@tas...
# prefect-community
j
I have a question about the task decorator
Copy code
@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?
n
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:
Copy code
@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
When moving to a non-development phase, you'll also want to take a look at Prefect Loggers, since
print
won't show up in your logs in Cloud/Server 🙂
j
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.
Copy code
@task
def read_s3(file,bucket_name, s3):
    obj = s3.Object(bucket_name, file_name)
    body = obj.get()['Body'].read()
yields
Invalid type for parameter Key, value: <Task: get_file_name>, type: <class 'prefect.tasks.core.function.FunctionTask'>, valid types: <class 'str'>
n
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:
Copy code
@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()
j
oh interesting, thanks!