I have a flow that runs locally but is returning a None result when run in Prefect Cloud. I think it's because I'm not setting the S3Result as the default? But this is confusing since I'm using the S3Storage as the default for flow storage. Does this seem right?

Are you printing inside the flow block?

Yes - the individual task is fairly plain, and seems to return a result with the local flow:

@task(name="Extract Owners from Streamline")
def extract_owners() -> DataFrame:
    """
    Grab Owners from Streamline and serialize as a Pandas DataFrame.
    """

    logger = prefect.context.get("logger")
    streamline = Streamline(token_key=TOKEN_KEY, token_secret=TOKEN_SECRET)
    owners_df: DataFrame = streamline.get_owners()

    if owners_df.empty:
        raise FAIL(
            "Owners DataFrame is empty. Something is wrong with the API pull from Streamline."
        )

    <http://logger.info|logger.info>(f"Pulled {owners_df.shape[0]} owners for {FLOW_NAME}")

    return owners_df

Wait, that thing you mentioned is more Prefect 2.0 I think

New flow run: I'm on Prefect 1-series still

What is the log of the shape?

Oh, interesting. S3Result forced it to produce .prefect_results after the task finished. I spoke too quickly.

This looks right. Are you sure the issue is not with

glue_database

?

No, glue-database is good. I think the current issue (separate from the original post) is now that the prefectTaskRole doesn't have access to the S3 bucket that I was testing locally to serialize the Parquet files too, but that's an easy fix to the IAM role.

The Result should not affect whether or not the return from a task is None