Yes. In Prefect 3.x you’ve got a few solid patterns to “fail” a task while still producing data you can use for an artifact. Here are the most common, with minimal examples: 1) Create the artifact inside the task, then raise - Artifacts are independent of task state. You can emit them and then fail.

from prefect import task
from prefect.artifacts import create_markdown_artifact

@task
def compute():
    try:
        # ... do work
        return {"ok": True}
    except Exception as exc:
        create_markdown_artifact(
            markdown=f"## Compute failed\n\n{exc}",
            key="compute-diagnostics",
            description="Failure diagnostics"
        )
        raise  # task will end in a Failed state, artifact still exists

2) Return a Failed state that includes your data, then read it without raising - You can attach any JSON-serializable payload to a Failed state and retrieve it with

raise_on_failure=False

.

from prefect import task, flow
from prefect.states import Failed
from prefect.artifacts import create_table_artifact

@task
def compute_partial():
    # ... partial work
    partial = {"rows": [{"id": 1, "status": "ok"}]}
    return Failed(message="Upstream validation failed", data=partial)

@flow
def main():
    fut = compute_partial.submit()
    partial = fut.result(raise_on_failure=False)  # no exception, returns the state's data
    create_table_artifact(table=partial["rows"], key="partial-output", description="Partial results")

3) Use an on_failure hook to create the artifact when the task fails - Hooks run when the task enters a Failed state; great for standardized failure output.

from prefect import task
from prefect.artifacts import create_markdown_artifact

def failure_artifact(task, task_run, state):
    create_markdown_artifact(
        markdown=f"Task {task.name} failed: {state.message}",
        key=f"failure-{task_run.id}"
    )

@task(on_failure=[failure_artifact])
def risky():
    raise ValueError("Boom")

Notes and limitations: - State data must be JSON-serializable. If you need richer content (tables, images, markdown), prefer emitting an artifact before failing. - To build artifacts downstream even if the upstream failed, call

future.result(raise_on_failure=False)

to access the data and avoid exceptions. - If you want downstream to run regardless of an upstream failure, build the artifact in the hook (pattern #3) or emit the artifact inside the failing task (pattern #1). Would you prefer to: - create the artifact inside the failing task, - or return data in a Failed state and build the artifact downstream?

Great question. In Prefect 3.x, the recommended way to cache binary API responses (like PDF bytes) is to use task result persistence + an appropriate serializer, and store the result in a filesystem or object store (S3/GCS/Azure). Artifacts are for UI metadata and aren’t designed to store raw binaries—use results for that. TL;DR recommended recipe - Enable result persistence on the task - Use a binary-friendly serializer (CompressedPickleSerializer) - Choose a storage backend (S3/GCS/Azure/local) via a block - Pick a cache policy (often INPUTS for URL-based caching) - Set a TTL via cache_expiration - For large files, set cache_result_in_memory=False to avoid keeping bytes in RAM Minimal example (S3 storage, cached by URL, 7-day TTL)

from datetime import timedelta
import requests
from prefect import task, flow
from prefect.serializers import CompressedPickleSerializer
from prefect.cache_policies import INPUTS

@task(
    persist_result=True,
    result_storage="s3-bucket/my-pdf-cache",  # Prefect block slug
    result_serializer=CompressedPickleSerializer(),
    cache_policy=INPUTS,                       # cache keyed by task inputs (e.g., URL)
    cache_expiration=timedelta(days=7),        # TTL
    cache_result_in_memory=False,              # don't keep large PDFs in RAM
)
def fetch_pdf(url: str) -> bytes:
    r = requests.get(url, timeout=30)
    r.raise_for_status()
    return r.content

@flow
def run():
    pdf_bytes = fetch_pdf("<https://example.com/report.pdf>")
    # On first run: fetches from API and persists to S3
    # Subsequent runs with same URL: loaded from cache/storage
    return pdf_bytes

Optional: custom cache key (e.g., ignore auth token differences)

import hashlib
from prefect.context import TaskRunContext

def url_only_cache_key(context: TaskRunContext, parameters: dict) -> str:
    return hashlib.sha256(parameters["url"].encode()).hexdigest()

@task(
    persist_result=True,
    result_storage="s3-bucket/my-pdf-cache",
    result_serializer=CompressedPickleSerializer(),
    cache_key_fn=url_only_cache_key,
)
def fetch_pdf_with_auth(url: str, token: str) -> bytes:
    r = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=30)
    r.raise_for_status()
    return r.content

Local filesystem instead of S3

@task(
    persist_result=True,
    result_storage="local-file-system/my-local-cache",  # LocalFileSystem block
    result_serializer=CompressedPickleSerializer(),
    cache_policy=INPUTS,
)
def fetch_pdf_local(url: str) -> bytes:
    return requests.get(url, timeout=30).content

Notes - Security/size: CompressedPickleSerializer reduces storage and bandwidth. If you don’t want pickle, you can write bytes to your own storage inside the task and return a pointer/path instead, but Prefect’s result persistence with Pickle/CompressedPickle is the simplest. - Integrations: For S3/GCS/Azure, install the corresponding collection and create a storage block (e.g., prefect-aws for S3). Then reference it by slug like

s3-bucket/my-pdf-cache

. - Artifacts vs results: Use results for the bytes (caching/persistence). You can still create artifacts for UI metadata (e.g., a link to the source URL), but don’t store binary blobs in artifacts. If you share which backend you prefer (S3/GCS/Azure/local), expected PDF sizes, and desired cache TTL, I can tailor a ready-to-paste snippet for your setup.

Yes—easily. In Prefect 3.x, dataclasses serialize cleanly with both JSON and Pickle serializers: - JSONSerializer: Works out of the box. Prefect’s default JSON object encoder leverages Pydantic’s to_jsonable_python, which understands dataclasses and converts them to/from dictionaries automatically. Round-trips preserve the dataclass type. - PickleSerializer: Also works without any extra config and supports arbitrary Python objects. Good for complex types; less human-readable. Quick examples - JSON (human-readable, portable)

from dataclasses import dataclass
from prefect import task, flow

@dataclass
class Product:
    id: int
    name: str
    price: float

@task(persist_result=True, result_serializer="json")
def get_product() -> Product:
    return Product(id=1, name="Widget", price=9.99)

@flow
def run():
    return get_product()

- Pickle (handles anything Pythonic)

from dataclasses import dataclass
from prefect import task

@dataclass
class Product:
    id: int
    name: str
    price: float

@task(persist_result=True, result_serializer="pickle")
def get_product_pickle() -> Product:
    return Product(id=2, name="Gadget", price=19.99)

- Large payloads: use compressed serializers to reduce storage size

@task(persist_result=True, result_serializer="compressed/json")
def get_large_dataclass() -> MyDataclass: ...
# or
@task(persist_result=True, result_serializer="compressed/pickle")
def get_large_dataclass_bin() -> MyDataclass: ...

When to pick which - Prefer JSON if you want readability, cross-language portability, or safer deserialization. - Prefer Pickle/CompressedPickle if your dataclass nests non-JSON-friendly types or you want max compatibility with Python objects. If your dataclass includes special types (e.g., datetime), JSON still works—either: - rely on Prefect’s default encoder/decoder (often fine), or - add a small custom encoder/decoder if you need precise formatting. If you share your dataclass shape (e.g., fields like datetime/Decimal/UUID), I can suggest the best serializer and any tiny encoder needed.

Short answer: - Pickle/CompressedPickle: Yes, BytesIO serializes and round-trips fine. - JSON: Not directly. Prefect’s JSON encoder treats IO streams as placeholders and does not read their content, which breaks deserialization. What happens with JSONSerializer - Prefect’s JSON serializer uses a custom encoder that detects io.IOBase (BytesIO, StringIO, etc.) and returns a placeholder object that explicitly says “original content not read.” - On load, the decoder can’t reconstruct a real BytesIO and will error. So JSON + BytesIO is a no-go. Recommended patterns 1) Use Pickle (or CompressedPickle) for full-fidelity streams

from prefect import task, flow
from prefect.serializers import CompressedPickleSerializer
import io

@flow(result_serializer=CompressedPickleSerializer())
def flow_with_stream():
    buf = io.BytesIO(b"important data")
    return buf  # round-trips fine

2) If you need JSON, convert to bytes/base64 yourself

import io, base64
from prefect import flow
from prefect.serializers import JSONSerializer

@flow(result_serializer=JSONSerializer())
def flow_json_safe():
    data = b"important binary data"
    encoded = base64.b64encode(data).decode("utf-8")
    return {"data_base64": encoded}

# later
# restored = io.BytesIO(base64.b64decode(result["data_base64"]))

3) Prefer plain bytes over BytesIO when using JSON

@flow(result_serializer="json")
def flow_bytes():
    return {"content": b"binary bytes"}  # bytes is JSON-encodable via Prefect’s encoder

When to choose which - Need fidelity and simplicity: Pickle/CompressedPickle - Need portability/JSON-only systems: base64 the bytes and return that - Avoid returning raw BytesIO with JSONSerializer If you want, tell me your storage backend and data sizes and I’ll suggest the best serializer plus example task config.