Got it! Assuming you’re on Prefect 3.x and you have a local “task storage” (persisted result) file, here’s a small script that will print the deserialized contents. It handles JSON, pickle/cloudpickle, and compressed variants. Save as

print_prefect_result.py

and run with:

python print_prefect_result.py /path/to/result-file

#!/usr/bin/env python3

import sys
import json
import base64
from pathlib import Path

def load_prefect_result(path: str):
    """Load and return the deserialized object from a Prefect result file."""
    p = Path(path).expanduser()
    raw = p.read_bytes()
    outer = json.loads(raw.decode("utf-8"))

    # Defensive defaults
    metadata = outer.get("metadata", {})
    serializer = (metadata.get("serializer") or {}).get("type")
    payload = outer.get("result")

    if serializer == "json":
        # Result is a JSON-encoded string
        return json.loads(payload)

    if serializer == "pickle":
        import cloudpickle  # Prefect depends on this; install if running standalone
        data = base64.decodebytes(payload.encode("utf-8"))
        return cloudpickle.loads(data)

    if serializer == "compressed/json":
        import lzma
        data = base64.decodebytes(payload.encode("utf-8"))
        decompressed = lzma.decompress(data).decode("utf-8")
        return json.loads(decompressed)

    if serializer == "compressed/pickle":
        import lzma, cloudpickle
        data = base64.decodebytes(payload.encode("utf-8"))
        decompressed = lzma.decompress(data)
        return cloudpickle.loads(decompressed)

    # Fallbacks if serializer metadata is missing or unknown
    # Try JSON-then-pickle heuristics
    if isinstance(payload, str):
        # Try JSON
        try:
            return json.loads(payload)
        except Exception:
            pass
        # Try base64+pickle
        try:
            import cloudpickle
            data = base64.decodebytes(payload.encode("utf-8"))
            return cloudpickle.loads(data)
        except Exception:
            pass

    # As a last resort, return the raw structure
    return outer

def main():
    if len(sys.argv) != 2:
        print("Usage: python print_prefect_result.py /path/to/result-file", file=sys.stderr)
        sys.exit(1)

    obj = load_prefect_result(sys.argv[1])

    # Pretty-print: JSON if possible, else repr
    try:
        print(json.dumps(obj, indent=2, default=str))
    except TypeError:
        # Non-JSON-serializable object; print repr
        from pprint import pprint
        pprint(obj)

if __name__ == "__main__":
    main()

Notes: - If you run this outside a Prefect environment, you may need

pip install cloudpickle

. - If your “storage file” is remote (e.g., on S3/GS), let me know and I can share a version that reads via fsspec/blocks. Quick check: are you on Prefect 3.x and is the file local?

Perfect — here’s a small, self-contained script you can run locally on Prefect 3.x to print the deserialized contents of a persisted result file. Usage: - Default local results live under

~/.prefect/storage/

(unless you’ve changed it). - To confirm your path:

prefect config view --show-defaults

and look for

PREFECT_RESULTS_LOCAL_STORAGE_PATH

. Save as

print_prefect_result.py

and run:

python print_prefect_result.py /path/to/result-file

#!/usr/bin/env python3

import sys
import json
import base64
from pathlib import Path

def load_prefect_result(path: str):
    """Load and return the deserialized object from a Prefect 3.x result file."""
    p = Path(path).expanduser()
    raw = p.read_bytes()
    outer = json.loads(raw.decode("utf-8"))

    # Metadata + payload
    metadata = outer.get("metadata", {})
    serializer = (metadata.get("serializer") or {}).get("type")
    payload = outer.get("result")

    if serializer == "json":
        return json.loads(payload)

    if serializer == "pickle":
        import cloudpickle
        data = base64.decodebytes(payload.encode("utf-8"))
        return cloudpickle.loads(data)

    if serializer == "compressed/json":
        import lzma
        data = base64.decodebytes(payload.encode("utf-8"))
        decompressed = lzma.decompress(data).decode("utf-8")
        return json.loads(decompressed)

    if serializer == "compressed/pickle":
        import lzma, cloudpickle
        data = base64.decodebytes(payload.encode("utf-8"))
        decompressed = lzma.decompress(data)
        return cloudpickle.loads(decompressed)

    # Heuristics if metadata missing/unknown
    if isinstance(payload, str):
        try:
            return json.loads(payload)
        except Exception:
            pass
        try:
            import cloudpickle
            data = base64.decodebytes(payload.encode("utf-8"))
            return cloudpickle.loads(data)
        except Exception:
            pass

    return outer

def main():
    if len(sys.argv) != 2:
        print("Usage: python print_prefect_result.py /path/to/result-file", file=sys.stderr)
        sys.exit(1)

    obj = load_prefect_result(sys.argv[1])

    # Pretty-print JSON if possible, else repr
    try:
        print(json.dumps(obj, indent=2, default=str))
    except TypeError:
        from pprint import pprint
        pprint(obj)

if __name__ == "__main__":
    main()

Notes: - Prefect 3.x defaults to storing local results under

~/.prefect/storage/

. - Persisted results must be enabled (globally via

prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

or per-`@flow`/`@task` with

persist_result=True

). - If you haven’t changed anything,

cloudpickle

should already be available via Prefect. If running outside,

pip install cloudpickle

may be needed. Want a version that, given a run ID, locates and prints the result automatically? I can share that too.

Awesome — here’s a script that takes a flow run ID or task run ID, fetches the latest (or final) state from your Prefect API, resolves the persisted result from local storage automatically, and prints it. It uses your current Prefect profile/connection (e.g.,

PREFECT_API_URL

) and works for Prefect 3.x local results (default

~/.prefect/storage/

). Usage: - Flow run:

python print_result_by_run_id.py --flow-run-id <UUID>

- Task run:

python print_result_by_run_id.py --task-run-id <UUID>

- Prefer final state (if available): add

--final

#!/usr/bin/env python3

import argparse
import asyncio
import json
import sys

from uuid import UUID
from prefect.client.orchestration import get_client

def choose_state(states, prefer_final: bool):
    if not states:
        return None
    if prefer_final:
        finals = [s for s in states if getattr(s, "is_final", lambda: False)()]
        if finals:
            # If multiple finals, pick the latest by timestamp if available
            try:
                finals.sort(key=lambda s: getattr(s, "timestamp", None) or 0)
            except Exception:
                pass
            return finals[-1]
    # Fallback: latest by position
    return states[-1]

async def fetch_state_and_result(flow_run_id: UUID = None, task_run_id: UUID = None, prefer_final: bool = False):
    async with get_client() as client:
        if flow_run_id:
            states = await client.read_flow_run_states(flow_run_id)
        else:
            states = await client.read_task_run_states(task_run_id)

        state = choose_state(states, prefer_final)
        if not state:
            raise RuntimeError("No states found for the given run.")

        # Try resolving the value via Prefect; this will read from local storage if persisted
        try:
            value = await state.aresult(raise_on_failure=False)
            return state, value
        except Exception as exc:
            # If result isn't persisted or can't be read, surface helpful info
            raise RuntimeError(f"Could not resolve result for the selected state: {exc}") from exc

def main():
    parser = argparse.ArgumentParser(description="Print Prefect 3.x result by run ID (flow or task).")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--flow-run-id", type=str, help="Flow run UUID")
    group.add_argument("--task-run-id", type=str, help="Task run UUID")
    parser.add_argument("--final", action="store_true", help="Prefer a final state if available")
    args = parser.parse_args()

    flow_run_id = UUID(args.flow_run_id) if args.flow_run_id else None
    task_run_id = UUID(args.task_run_id) if args.task_run_id else None

    try:
        state, result = asyncio.run(
            fetch_state_and_result(flow_run_id=flow_run_id, task_run_id=task_run_id, prefer_final=args.final)
        )
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(2)

    # Print some context
    print(f"State: {state.type.name if hasattr(state, 'type') else state}")
    if getattr(state, "message", None):
        print(f"Message: {state.message}")

    # Pretty-print result: JSON if possible, else repr
    print("Result:")
    try:
        print(json.dumps(result, indent=2, default=str))
    except TypeError:
        # Non-JSON-serializable object
        from pprint import pprint
        pprint(result)

if __name__ == "__main__":
    main()

Notes: - This will work whether the result is embedded or persisted; Prefect’s

state.aresult()

handles deserialization and reading from local storage using the metadata on the state. - If you see an error like “Could not resolve result…”, it usually means the result wasn’t persisted (e.g.,

persist_result=False

), the file expired/was moved, or the script is running on a different machine than where the result was stored. - You can prefer a final state with

--final

. Otherwise, it uses the latest state returned by the API.