I'm getting FlowStorageError when using the SnowflakeQueriesFromFile task - how to fix it?

Archive
1

This error typically occurs when your agent doesn’t have access to the query file. Make sure that the SQL query file is available on the agent’s machine.

Here is an example for using this 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()

Alternatively, you can switch to the SnowflakeQuery task which loads your query from a string.