Got it — missing logs in Cloud usually come down to buffering/flush timing, context, or log level in Prefect 2.x. Here’s a short checklist to make logs from

prefect.get_run_logger()

show up reliably: Quick fixes to try first - Ensure API logging is on and level is not too high - Set:

PREFECT_LOGGING_TO_API_ENABLED=true
    PREFECT_LOGGING_LEVEL=INFO

- If you expect DEBUG logs, use

PREFECT_LOGGING_LEVEL=DEBUG

. - Reduce batching delay so logs are shipped more often (helps if workers exit quickly)

PREFECT_LOGGING_TO_API_BATCH_INTERVAL=0.5  # default is 2.0s

- Avoid oversize messages (single logs > ~1MB are dropped) - Split very large payloads into multiple

<http://logger.info|logger.info>(...)

calls. - Capture prints (very reliable across environments, including Dask/Ray/K8s) - Add `log_prints=True`:

@flow(log_prints=True)
    @task(log_prints=True)

Flush logs when runs may exit/crash - Logs are batched and sent asynchronously. If a process exits or crashes, buffered logs can be lost. Add an explicit flush in exception paths and on shutdown: Sync flows/tasks:

from prefect import flow, get_run_logger
from prefect.logging.handlers import APILogHandler

@flow
def my_flow():
    logger = get_run_logger()
    try:
        # your work here
        ...
    except Exception:
        logger.exception("Flow failed")
        APILogHandler.flush()  # ensure errors are sent
        raise
    finally:
        APILogHandler.flush()

Async flows/tasks:

from prefect import flow, get_run_logger
from prefect.logging.handlers import APILogHandler

@flow
async def my_flow():
    logger = get_run_logger()
    try:
        ...
    except Exception:
        logger.exception("Flow failed")
        await APILogHandler.aflush()
        raise
    finally:
        await APILogHandler.aflush()

Context gotchas - Only call

get_run_logger()

inside an active flow/task. If you call it: - before the flow starts, - in a background thread/process without Prefect context, - or after the flow/task has finished, the log will be dropped (or raise a missing context error). - For background threads/processes, create logs inside the task/flow where the context exists, or pass data back and log in the main task/flow thread. Distributed runners and GCP environments - If you’re using Dask/Ray/Kubernetes (GKE) workers, the worker process may end before logs flush. The fixes above help: - Lower

PREFECT_LOGGING_TO_API_BATCH_INTERVAL

- Add explicit flushes when tasks are likely to raise/exit - Prefer

log_prints=True

to capture stdout reliably - Avoid `os._exit()`/hard kills that skip cleanup; raise Exceptions instead Extra knobs if you use other libraries - If you want 3rd-party library logs (e.g.,

google-cloud-*

) in Prefect Cloud, add them:

PREFECT_LOGGING_EXTRA_LOGGERS=google,google.cloud

A couple of quick questions to pinpoint your case - Where are the logs missing: at the end of runs, around crashes, or randomly during normal operation? - How are you running: local agent/worker, Docker, GKE, Cloud Run, Dask/Ray task runner? - Are you using

print()

as well, and if so, are those missing too? - Do you see DEBUG logs missing (level issue) or INFO/ERROR logs (likely buffering/flush or context)? If you can share a run ID and how you execute (e.g., GKE + DaskTaskRunner), I can recommend the most targeted fix.

Thanks for flagging this. In Prefect Cloud (and you mentioned you’re on Prefect 2 + Cloud), that error usually means a string you’re sending to the Cloud API exceeds a managed limit of 25,000 characters. The most common culprits are: - A log message that’s too long (e.g., logging a huge dict/DataFrame/JSON) - A state message that’s too long (e.g., setting

Failed(message=...)

or raising an exception with a giant string) This 25k limit is enforced by the Cloud API (it’s not defined in the open-source Prefect 2 schemas), so anything over that length will be rejected with “String should have at most 25000 characters.” Quick ways to fix it - Truncate before sending to Cloud:

MAX = 25000

  def truncate(s, n=MAX):
      s = str(s)
      return s if len(s) <= n else s[: n - 12] + "… [truncated]"

- Truncate log messages:

from prefect import flow, get_run_logger

  MAX = 25000

  def truncate(s, n=MAX):
      s = str(s)
      return s if len(s) <= n else s[: n - 12] + "… [truncated]"

  @flow
  def demo():
      logger = get_run_logger()
      big = "x" * 30001
      <http://logger.info|logger.info>(truncate(big))

  if __name__ == "__main__":
      demo()

- Truncate state messages:

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

  @task
  def do_work():
      big_payload = "x" * 30001
      return Failed(message=truncate(big_payload))

- Store large details externally and log a link: - Since you’re on GCP, upload the full text to GCS and reference the

gs://

URL in the log/state message.

from google.cloud import storage

  def upload_text_to_gcs(bucket, path, text):
      client = storage.Client()
      blob = client.bucket(bucket).blob(path)
      blob.upload_from_string(text, content_type="text/plain")
      return f"gs://{bucket}/{path}"

Then:

try:
      ...
  except Exception:
      details = "very long text..."
      url = upload_text_to_gcs("my-bucket", "errors/run-123.txt", details)
      raise RuntimeError(f"Error details stored at {url}")

Helpful docs (Prefect 2.x) - Logs: Logs - States: States - Artifacts (for richer attachments/links): Artifacts Can you share where this is happening in your run? - Is it triggered by a specific

<http://logger.info|logger.info>(...)

or

logger.error(...)

? - Or when you set or return a state with a large

message

? - A snippet or the traceback would help me point you to the exact spot to truncate or offload to GCS.

Great question. In Prefect 2 (Cloud), artifacts are the right way to persist rich data (tables, markdown, links) without hitting log truncation. They’re rendered in the UI and can be versioned over time with a key. Quick examples you can drop into your flows/tasks: - Persist a long string/report (markdown)

from prefect import flow
from prefect.artifacts import create_markdown_artifact

@flow
def my_flow():
    report = """# Model Results

- Accuracy: 0.982
- F1: 0.975

Full confusion matrix and notes below...
"""
    create_markdown_artifact(
        key="daily-model-report",          # use a stable key to version over runs
        markdown=report,
        description="Daily model evaluation report"
    )

my_flow()

- Persist structured data as a table (e.g., instead of logging a big JSON/df head)

from prefect import flow
from prefect.artifacts import create_table_artifact

@flow
def my_flow():
    rows = [
        {"id": 1, "score": 0.98, "label": "A"},
        {"id": 2, "score": 0.76, "label": "B"},
    ]
    create_table_artifact(
        key="scoring-summary",
        table=rows,                        # can also be dict-of-lists or list-of-lists
        description="Latest scoring summary"
    )

my_flow()

- Persist a link to large data in GCS (recommended for big payloads)

from prefect import flow
from prefect.artifacts import create_link_artifact

@flow
def my_flow():
    # Upload your file to GCS however you prefer, then link to it.
    # Example GCS URI or signed URL:
    gcs_url = "<https://storage.googleapis.com/<bucket>/<path-to-object>>"

    create_link_artifact(
        key="scoring-output-location",
        link=gcs_url,
        link_text="Open GCS output",
        description="Full scoring output stored in GCS"
    )

my_flow()

What you’ll see in Prefect Cloud - Artifacts appear under the flow run’s Artifacts tab. - If you set a

key

, the latest version also shows up on the workspace-wide Artifacts page. Reusing the same key across runs gives you a history/versioning view. - Artifacts render nicely (markdown, tables, clickable links), which avoids log truncation issues. Tips and constraints - Always set a

key

if you want the artifact to show on the global Artifacts page. - Keep data JSON-serializable. For very large datasets, store them in GCS and publish a link artifact. - Table artifacts auto-sanitize NaN to null for JSON compatibility. - Use consistent keys (e.g., “daily-model-report”) to get a versioned timeline across runs. Docs and references - Concepts: Artifacts (Prefect 2) - Python API: prefect.artifacts API If you share what kind of data you’re logging today (e.g., a pandas DataFrame, a big JSON payload, model metrics), I can tailor a drop-in artifact snippet for your flow—and a GCS pattern if you want to persist the full file and link to it.

You’ve got two good options in Prefect 2, depending on how large the JSON is: 1) Inline it as a Markdown artifact (pretty-printed with syntax highlighting) - Best for moderate-sized JSON you want visible directly in the Prefect UI.

from prefect import flow
from prefect.artifacts import create_markdown_artifact
import json

@flow
def save_pretty_json_inline(data: dict, key: str = "pretty-json"):
    pretty = json.dumps(data, indent=2, sort_keys=True)
    md = f"""### JSON Output

```json
{pretty}

""" create_markdown_artifact( key=key, # set a stable key to get versioning on the Artifacts page markdown=md, description="Pretty-formatted JSON" ) # Example # save_pretty_json_inline({"a": 1, "b": {"c": 2}})

2) Store the JSON in GCS and publish a link artifact
- Recommended for large payloads (keeps the UI snappy and avoids payload limits).
- This uses a signed URL so it’s a one-click download from Prefect Cloud.

from prefect import flow from prefect.artifacts import create_link_artifact from google.cloud import storage from datetime import timedelta import json import uuid @flow def save_json_to_gcs_and_link(data: dict, bucket_name: str, key: str = "pretty-json"): client = storage.Client() bucket = client.bucket(bucket_name) # Unique-ish path per run; adjust as needed blob_name = f"artifacts/{uuid.uuid4()}/result.json" blob = bucket.blob(blob_name) blob.upload_from_string( json.dumps(data, indent=2, sort_keys=True), content_type="application/json" ) # Create a time-limited signed URL (adjust expiration as needed) url = blob.generate_signed_url( version="v4", expiration=timedelta(days=7), method="GET" ) create_link_artifact( key=key, link=url, link_text="Download pretty JSON", description=f"GCS object: gs://{bucket_name}/{blob_name}" ) # Example # save_json_to_gcs_and_link({"a": 1, "b": {"c": 2}}, bucket_name="my-bucket") ``` Notes - Always set a

key

so the artifact appears on the workspace-wide Artifacts page and versions across runs. - For very large JSON, prefer the GCS + link artifact approach to avoid large payloads in the artifact itself. - Use

create_markdown_artifact

when you want the content rendered inline in Prefect Cloud. Docs - Concepts: Artifacts (Prefect 2) - API: prefect.artifacts API If you share roughly how big your JSON is and whether you want it inline or downloadable, I can tailor the example further (e.g., naming, foldering in GCS, signed URL duration).

Short answer: there isn’t a published, fixed TTL for artifacts in Prefect Cloud 2. They generally persist until you delete them. However, Cloud does enforce retention for other objects (like flow runs) based on your plan, and when those runs are pruned, artifacts tied to them may no longer be visible in the UI. Prefect does not publicly document a separate, time-based retention window specifically for artifacts. What this means in practice: - If you need long-term retention, treat artifacts as UI/summary objects and store the large payload in your own storage (e.g., GCS) and publish a link artifact. - If your account has a flow run retention window, older runs (and their context in the UI) may be pruned according to your plan. Artifacts are not documented with a different retention policy. Official references: - Artifacts overview: Artifacts (Prefect 2) - Cloud retention note (applies to flow runs): Flow run retention in Cloud If you share your plan tier (Team/Enterprise) I can check what it implies for your workspace, or I can loop in Support for the exact retention behavior on your account.

Great question — they serve different purposes. - Artifacts (Prefect 2 and 3) - What they are: Human-readable outputs persisted and rendered in the UI (markdown, tables, links; in 3.x also images/progress). - Typical use: Dashboards/reports, pretty JSON, sample rows, links to files in GCS/S3, QA summaries. Versioned over time when you set a

key

. - Where you see them: On run detail pages and on the global Artifacts page (if a

key

is set). - API:

prefect.artifacts.create_markdown_artifact

,

create_table_artifact

,

create_link_artifact

, or

Artifact

class. - Docs (Prefect 2): Artifacts (concepts) • Artifacts API - Assets (Prefect 3 only) - What they are: A way to model actual data objects your flows produce/consume (e.g., s3://..., snowflake://...), with lineage and rich metadata (owners, URLs). - Typical use: Data lineage graphs, ownership/governance, dependency tracking between datasets, asset catalog in the UI. - API:

prefect.assets.Asset

,

@materialize(...)

,

add_asset_metadata(...)

. - Docs (Prefect 3): Assets (guide) What should you use? - You’re on Prefect 2 Cloud: use artifacts. For big JSON, either: - Inline it as a markdown artifact (pretty-printed) if it’s moderate in size and you want it visible in the UI. - Store the JSON in GCS and publish a link artifact for very large payloads (recommended for GCP workloads). - If you need an asset catalog, lineage graphs, owners, and dependency tracking, that’s a Prefect 3 “Assets” feature—worth considering an upgrade. If you share whether you need lineage/ownership vs. just better visibility and persistence for outputs, I can recommend a concrete path (and provide examples tailored for GCS).

Great question. In unit tests you generally don’t want

create_markdown_artifact()

to hit the Prefect API. In Prefect 2 it will try to create an artifact even outside a flow/task context (no error), so the safest approach is to mock it in tests. Here are solid patterns you can use: 1) Patch where you import it (most common) - Patch the symbol in your module, not in

prefect.artifacts

, so your production code stays untouched and no network calls are made.

# my_module.py
from prefect.artifacts import create_markdown_artifact

def publish_report(data: dict) -> None:
    md = f"# Report\n\nItems: {len(data)}"
    create_markdown_artifact(key="report", markdown=md)

# test_my_module.py
from unittest.mock import patch
from my_module import publish_report

def test_publish_report_makes_artifact():
    with patch("my_module.create_markdown_artifact") as mock_create:
        publish_report({"a": 1, "b": 2})
        mock_create.assert_called_once()
        args, kwargs = mock_create.call_args
        assert kwargs["key"] == "report"
        assert "Items: 2" in kwargs["markdown"]

2) Patch the Prefect client to assert payloads (deeper test, still no network) - Useful if you want to validate what would be sent to the API.

from unittest.mock import patch, AsyncMock
from prefect.artifacts import create_markdown_artifact

def test_create_markdown_artifact_calls_client():
    with patch("prefect.client.utilities.get_or_create_client") as get_client:
        fake_client = AsyncMock()
        fake_client.create_artifact = AsyncMock(return_value="uuid-123")
        get_client.return_value = (fake_client, False)

        create_markdown_artifact(key="k", markdown="# hi")

        fake_client.create_artifact.assert_awaited_once()
        # Inspect the request payload
        args, kwargs = fake_client.create_artifact.call_args
        artifact = args[0]
        assert artifact.data["markdown"] == "# hi"
        assert artifact.key == "k"

3) Add a tiny wrapper you can toggle off in tests (feature flag) - Keeps your business logic testable while making artifact emission optional in test/CI.

# artifacts_helpers.py
import os
from prefect.artifacts import create_markdown_artifact

def maybe_create_markdown(key: str, md: str, description: str | None = None):
    if os.getenv("DISABLE_PREFECT_ARTIFACTS") == "1":
        return None
    return create_markdown_artifact(key=key, markdown=md, description=description)

# test_my_module.py
from unittest.mock import patch
import os

def test_disabled_artifacts_skip_api(monkeypatch):
    monkeypatch.setenv("DISABLE_PREFECT_ARTIFACTS", "1")
    with patch("artifacts_helpers.create_markdown_artifact") as mock_create:
        from artifacts_helpers import maybe_create_markdown
        maybe_create_markdown("k", "# hi")
        mock_create.assert_not_called()

4) Light integration test with Prefect’s test harness - If you want to exercise a small flow and ensure the call returns an ID.

from prefect import flow
from prefect.artifacts import create_markdown_artifact
from prefect.testing.utilities import prefect_test_harness

def test_flow_creates_artifact():
    with prefect_test_harness():
        @flow
        def f():
            return create_markdown_artifact(key="k", markdown="# hi")

        artifact_id = f()
        assert artifact_id  # ID returned by API

Notes and gotchas - Outside a flow/task: in Prefect 2,

create_markdown_artifact()

does not raise; it will still try to call the API (with no run IDs). That’s why patching in unit tests is recommended. - No API available: network/API errors will propagate. Avoid in unit tests by patching. - Patch the right name: patch where your code imported the function (e.g.,

my_module.create_markdown_artifact

), not

prefect.artifacts.create_markdown_artifact

. Docs - Artifacts (Prefect 2 concepts): Artifacts

- Artifacts API: prefect.artifacts API If you share a small snippet of the function under test, I can show an exact patch path and a test tailored to your setup (pytest/unittest, with or without CI env vars).