@Marvin Why am I getting the following error after my subflow deployment successfully completes?

prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

My code:

import argparse
from prefect import flow, task, get_run_logger
import prefect.deployments
import utils

DATETIME = utils.get_current_timestamp()
FLOW_RUN_NAME = "il-workflow-{plate_name}-" + DATETIME


@task
def run_curation(
    plate_name: str, bfx_run_label: str, bio_exp_id: int, dev: bool = False
):
    flow_run = prefect.deployments.run_deployment(
        name=f'ild-curation/ild-curation-{"dev" if dev else "prod"}',
        parameters={
            "plate_name": plate_name,
            "bfx_run_label": bfx_run_label,
            "bio_exp_id": bio_exp_id,
            "quilt_s3_bucket": f'<s3://compbio-curated>-{"dev" if dev else "prod"}',
        },
    )
    flow_run.state.result()


@flow(flow_run_name=FLOW_RUN_NAME)
def ild_workflow(
    plate_name: str, bfx_run_label: str, bio_exp_id: int, dev: bool = False
):
    logger = get_run_logger()
    logger.info(f'Running {"dev" if dev else "prod"} ILD workflow for {plate_name=}...')

    logger.info("Launching curation...")
    run_curation(plate_name, bfx_run_label, bio_exp_id, dev)

Your message was too long by 39 tokens - please shorten it and try again. For reference, here's your message at the allowed limit:

Why am I getting the following error after my subflow deployment successfully completes?

prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

My code: ```import argparse from prefect import flow, task, get_run_logger import prefect.deployments import utils DATETIME = utils.get_current_timestamp() FLOW_RUN_NAME = "il-workflow-{plate_name}-" + DATETIME @task def run_curation( plate_name: str, bfx_run_label: str, bio_exp_id: int, dev: bool = False ): flow_run = prefect.deployments.run_deployment( name=f'ild-curation/ild-curation-{"dev" if dev else "prod"}', parameters={ "plate_name": plate_name, "bfx_run_label": bfx_run_label, "bio_exp_id": bio_exp_id, "quilt_s3_bucket": f's3://compbio-curated-{"dev" if dev else "prod"}', }, ) flow_run.state.result() @flow(flow_run_name=FLOW_RUN_NAME) def ild_workflow( plate_name: str, bfx_run_label: str, bio_exp_id: int, dev: bool = False ): logger = get_run_logger() logger.info(f'Running {"dev" if dev else "prod

does the subflow you're running as a deployment have result persistence enabled?

I don't think so, not aware of this concept!

yep! if you're accessing

flow_run.state.result()

, prefect is going to try to fetch a result file containing the return value of the flow function

❯ prefect config set PREFECT_DEFAULT_RESULT_STORAGE_BLOCK='cyflows-result-storage'
Set 'PREFECT_DEFAULT_RESULT_STORAGE_BLOCK' to 'cyflows-result-storage'.
Updated profile 'default'.
❯ prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
Set 'PREFECT_RESULTS_PERSIST_BY_DEFAULT' to 'true'.
Updated profile 'default'.

I just reran and it seems the same thing happened. Is there anything else I need to do to make sure my flow runs pick this up?

are those deployments running locally? or in ECS? if in a remote env, you'll need to set those as env vars on the work pool or deployment

ECS fargate! push pool