<@ULVA73B9P> I've enabled prefect result persistan...
# ask-marvin
r
@Marvin I've enabled prefect result persistance on s3, I want to purge my flow run's data hence also persistence data. How can I do this ?
m
thought for 150.0 seconds
Short answer: Prefect does not automatically purge persisted results from S3 when you delete a flow run. So if you want to remove both: 1. Prefect metadata for the flow run, and 2. the actual persisted result objects in S3, you need to handle the S3 deletion yourself. What happens when you delete a flow run - Deleting a flow run in Prefect removes the run record/state metadata from Prefect - It does not delete the corresponding persisted result files from your S3 bucket So S3 cleanup is a separate step. How Prefect stores persisted results When
persist_result=True
is enabled, Prefect writes results to the configured result storage and stores a storage key in the flow/task state metadata. For S3-backed result storage, that key maps to the object in your bucket. In practice, purging a run’s persisted data means: - read the flow run’s states - extract any
storage_key
values from persisted results - delete those objects from S3 Recommended approach If you need per-flow-run cleanup, write a small cleanup script/flow that: 1. Reads the flow run states 2. Collects persisted result `storage_key`s 3. Deletes those keys from S3 with
boto3
4. Optionally deletes the flow run metadata from Prefect afterward Example:
Copy code
python
from uuid import UUID
import boto3
from prefect.client.orchestration import get_client

async def purge_flow_run_results_from_s3(flow_run_id: UUID, bucket: str, prefix: str = ""):
    s3 = boto3.client("s3")
    deleted = []

    async with get_client() as client:
        states = await client.read_flow_run_states(flow_run_id)

        for state in states:
            data = getattr(state, "data", None)
            storage_key = getattr(data, "storage_key", None)

            if not storage_key:
                continue

            s3_key = f"{prefix}/{storage_key}".lstrip("/")
            s3.delete_object(Bucket=bucket, Key=s3_key)
            deleted.append(s3_key)

    return deleted
If your result storage uses a bucket/prefix layout, adjust
prefix
accordingly. Important caveat The exact S3 object path depends on how your result storage is configured. The key thing to inspect is
state.data.storage_key
. Best long-term option If your main goal is retention management, the simplest solution is usually an S3 lifecycle policy on the bucket/prefix used for Prefect results. That way old result objects are expired automatically without needing custom cleanup code. AWS docs: - S3 lifecycle policies Related Prefect docs - Prefect results - Result persistence If you want, I can help you with either: - a full cleanup script that deletes both Prefect flow-run metadata and S3 results, or - an S3 lifecycle rule for your Prefect result prefix.