Hi everyone, I'm trying to use "SnowflakeQueriesFromFile" in a task and I'm getting a strange error. Not sure how to start debugging this, any ideas? Could it be something to do with the file path?

"Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n NameError("name \'err\' is not defined")')"

The task itself is straightforward, it looks like this:

snowsql_obj = SnowflakeQueriesFromFile(account=SNOWFLAKE_ACCOUNT, user=SNOWFLAKE_USER, password=SNOWFLAKE_PWD, file_path="../../sql/amplitude_raw.sql")
snowsql_obj.run()

You issue is because of flow storage. This thread may help you. You need to make sure your agent can access both your flow file and your query file. An absolute path would be better than relative (atm you use a relative path). If you need an example for the task:

from prefect import Flow, Parameter, task
from prefect.tasks.snowflake import SnowflakeQueriesFromFile
from prefect.tasks.secrets import PrefectSecret


snowflake_query = SnowflakeQueriesFromFile(
    database="DEV", schema="jaffle_shop", autocommit=True, file_path="sample_query.sql"
)


@task(log_stdout=True)
def print_data(x):
    print(x)  # prints a list of tuples


with Flow("snowflake_tasks_example") as flow:
    account = PrefectSecret("SNOWFLAKE_ACCOUNT_ID")
    user = PrefectSecret("SNOWFLAKE_USER")
    pwd = PrefectSecret("SNOWFLAKE_PASS")
    role = PrefectSecret("SNOWFLAKE_ROLE")
    warehouse = Parameter("SNOWFLAKE_WAREHOUSE", default="COMPUTE_WH")
    data = snowflake_query(
        account=account, user=user, password=pwd, role=role, warehouse=warehouse,
    )
    print_data(data)

if __name__ == "__main__":
    flow.run()

Thanks! Thats very helpful. So, I know that my flow storage can be accesssed (because I can run the flow properly without the snowflake task). I guess this means that the sql file cannot be accessed. Since I'm using an S3 bucket for the storage, do I need to place the SQL file in the same s3 bucket?

No, it must be placed on the agent. Afaik this task expects a local file, so S3 path won’t work. Perhaps you can add a task before that and download this query file from S3?

Thanks for the info, I will check out the resources. Much appreciated!