python(fix): disable backups by default for sift_client ingestion

python/changelog.d/+ingestion-backups-opt-in.breaking.md

@@ -0,0 +1 @@
+Ingestion streaming clients now default to `StreamingMode.LIVE_ONLY`; pass `streaming_mode=StreamingMode.LIVE_WITH_BACKUPS` to enable local telemetry backups.

python/docs/examples/ingestion.ipynb

@@ -17,6 +17,12 @@
     "pip install sift-stack-py[sift-stream]\n",
     "```\n",
     "\n",
+    "## Streaming backups\n",
+    "\n",
+    "Both streaming client factories default to `StreamingMode.LIVE_ONLY`: telemetry streams to Sift without local disk backups.\n",
+    "\n",
+    "To enable local backups and replay after delivery failures, import `StreamingMode` from `sift_client.resources.ingestion` and pass `streaming_mode=StreamingMode.LIVE_WITH_BACKUPS` to `create_ingestion_config_streaming_client` or `create_auto_register_streaming_client`.\n",
+    "\n",
     "## Topics Covered\n",
     "\n",
     "1. **Basic Example**: Simple single flow sending using FlowConfig\n",

python/examples/ingestion-tutorial/stream.py

@@ -170,6 +170,10 @@ async def main() -> None:
     # All telemetry sent within this context will appear inside
     # this Run in Sift.
     #
+    # Streaming defaults to LIVE_ONLY (no local telemetry backups).
+    # To opt in, import StreamingMode from sift_client.resources.ingestion
+    # and pass streaming_mode=StreamingMode.LIVE_WITH_BACKUPS below.
+    #
     # WARNING: SiftStream should be reused for the entire program duration.
     # Do NOT recreate the streaming client per batch or per iteration —
     # each instantiation creates a new ingestion config, which introduces

python/lib/sift_client/_internal/low_level_wrappers/ingestion.py

@@ -278,7 +278,7 @@ async def _build_sift_stream_instance(
     _default_logfile_prefix = "sift_stream_bindings.log"
 
     if streaming_mode is ...:  # type: ignore[comparison-overlap]
-        streaming_mode = StreamingMode.LIVE_WITH_BACKUPS
+        streaming_mode = StreamingMode.LIVE_ONLY
 
     # None disables metrics streaming, so it passes through to the builder as-is. Zero is
     # rejected here because SiftStream cannot schedule metrics on a zero interval.
@@ -328,7 +328,7 @@ async def _build_sift_stream_instance(
             fb_builder.disk_backup_policy = disk_backup_policy
         return await fb_builder.build()
 
-    else:  # LIVE_WITH_BACKUPS (default)
+    else:  # LIVE_WITH_BACKUPS
         lwb_builder = config_builder.live_with_backups()
         lwb_builder.metrics_streaming_interval = metrics_interval
         if retry_policy is not None:

python/lib/sift_client/_tests/_internal/low_level_wrappers/test_ingestion_stream_builder.py

@@ -41,6 +41,36 @@ async def _build(**kwargs):
     )
 
 
+class TestStreamingMode:
+    @pytest.mark.asyncio
+    async def test_defaults_to_live_only(self, config_builder):
+        await _build()
+
+        config_builder.live_only.return_value.build.assert_awaited_once()
+        config_builder.live_with_backups.assert_not_called()
+        config_builder.file_backup.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_backups_are_explicit_opt_in(self, config_builder):
+        backup_policy = sift_stream_bindings.DiskBackupPolicyPy.default()
+        retry_policy = sift_stream_bindings.RetryPolicyPy.default()
+
+        await _build(
+            streaming_mode=StreamingMode.LIVE_WITH_BACKUPS,
+            disk_backup_policy=backup_policy,
+            retry_policy=retry_policy,
+            checkpoint_interval_seconds=30,
+        )
+
+        mode_builder = config_builder.live_with_backups.return_value
+        mode_builder.build.assert_awaited_once()
+        assert mode_builder.disk_backup_policy is backup_policy
+        assert mode_builder.retry_policy is retry_policy
+        assert mode_builder.checkpoint_interval.to_timedelta() == timedelta(seconds=30)
+        config_builder.live_only.assert_not_called()
+        config_builder.file_backup.assert_not_called()
+
+
 @pytest.mark.parametrize("streaming_mode", list(MODE_BUILDER_METHODS))
 class TestMetricsStreamingInterval:
     @pytest.mark.asyncio

python/lib/sift_client/_tests/resources/test_ingestion.py

@@ -1,4 +1,4 @@
-"""Tests that rate-limit errors from SiftStream reach callers of the streaming clients unchanged."""
+"""Tests for streaming client configuration and rate-limit error propagation."""
 
 from datetime import timedelta
 from unittest.mock import AsyncMock, MagicMock
@@ -7,12 +7,45 @@
 
 from sift_client.resources.ingestion import (
     AutoRegisterStreamingClient,
+    IngestionAPIAsync,
     IngestionConfigStreamingClient,
+    StreamingMode,
 )
 
 sift_stream_bindings = pytest.importorskip("sift_stream_bindings")
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "factory_name",
+    ["create_ingestion_config_streaming_client", "create_auto_register_streaming_client"],
+)
+@pytest.mark.parametrize("streaming_mode", [None, StreamingMode.LIVE_WITH_BACKUPS])
+async def test_streaming_mode_default_and_opt_in(
+    mock_client, monkeypatch, factory_name, streaming_mode
+):
+    build_stream = AsyncMock()
+    monkeypatch.setattr(
+        "sift_client._internal.low_level_wrappers.ingestion._build_sift_stream_instance",
+        build_stream,
+    )
+    monkeypatch.setattr("sift_client.resources.ingestion._build_sift_stream_instance", build_stream)
+    monkeypatch.setattr(
+        sift_stream_bindings,
+        "SiftStreamAutoRegisterPy",
+        MagicMock(from_stream=AsyncMock()),
+    )
+    api = IngestionAPIAsync(mock_client)
+    kwargs = {} if streaming_mode is None else {"streaming_mode": streaming_mode}
+
+    await getattr(api, factory_name)(ingestion_config=MagicMock(), **kwargs)
+
+    build_stream.assert_awaited_once()
+    assert build_stream.call_args.kwargs["streaming_mode"] == (
+        StreamingMode.LIVE_ONLY if streaming_mode is None else streaming_mode
+    )
+
+
 def _rate_limited_error():
     return sift_stream_bindings.RateLimitedErrorPy(
         "[RateLimitedError] rate limited",

python/lib/sift_client/resources/ingestion.py

@@ -159,7 +159,7 @@ async def create_ingestion_config_streaming_client(
         run: RunCreate | dict | str | Run | None = None,
         asset_tags: list[str] | list[Tag] | None = None,
         asset_metadata: dict[str, str | float | bool] | None = None,
-        streaming_mode: StreamingMode = StreamingMode.LIVE_WITH_BACKUPS,
+        streaming_mode: StreamingMode = StreamingMode.LIVE_ONLY,
         retry_policy: RetryPolicyPy | None = None,
         disk_backup_policy: DiskBackupPolicyPy | None = None,
         checkpoint_interval_seconds: int | None = None,
@@ -175,7 +175,8 @@ async def create_ingestion_config_streaming_client(
             run: The run to associate with ingestion. Can be a Run, RunCreate, dict, or run ID string.
             asset_tags: Tags to associate with the asset.
             asset_metadata: Metadata to associate with the asset.
-            streaming_mode: Transport mode for the stream. Defaults to LIVE_WITH_BACKUPS.
+            streaming_mode: Transport mode for the stream. Defaults to LIVE_ONLY, without disk
+                backups. Set to LIVE_WITH_BACKUPS to opt in to local disk backups.
             retry_policy: Retry policy for LIVE_WITH_BACKUPS mode.
             disk_backup_policy: Disk backup policy for LIVE_WITH_BACKUPS or FILE_BACKUP mode.
             checkpoint_interval_seconds: Checkpoint interval in seconds (LIVE_WITH_BACKUPS only).
@@ -223,7 +224,7 @@ async def create_auto_register_streaming_client(
         run: RunCreate | dict | str | Run | None = None,
         asset_tags: list[str] | list[Tag] | None = None,
         asset_metadata: dict[str, str | float | bool] | None = None,
-        streaming_mode: StreamingMode = StreamingMode.LIVE_WITH_BACKUPS,
+        streaming_mode: StreamingMode = StreamingMode.LIVE_ONLY,
         retry_policy: RetryPolicyPy | None = None,
         disk_backup_policy: DiskBackupPolicyPy | None = None,
         checkpoint_interval_seconds: int | None = None,
@@ -256,7 +257,8 @@ async def create_auto_register_streaming_client(
             run: The run to associate with ingestion. Can be a Run, RunCreate, dict, or run ID string.
             asset_tags: Tags to associate with the asset.
             asset_metadata: Metadata to associate with the asset.
-            streaming_mode: Transport mode for the stream. Defaults to LIVE_WITH_BACKUPS.
+            streaming_mode: Transport mode for the stream. Defaults to LIVE_ONLY, without disk
+                backups. Set to LIVE_WITH_BACKUPS to opt in to local disk backups.
             retry_policy: Retry policy for LIVE_WITH_BACKUPS mode.
             disk_backup_policy: Disk backup policy for LIVE_WITH_BACKUPS or FILE_BACKUP mode.
             checkpoint_interval_seconds: Checkpoint interval in seconds (LIVE_WITH_BACKUPS only).
@@ -327,7 +329,7 @@ async def _create(
         run: RunCreate | dict | str | Run | RunFormPy | None = None,
         asset_tags: list[str] | list[Tag] | None = None,
         asset_metadata: dict[str, str | float | bool] | None = None,
-        streaming_mode: StreamingMode = StreamingMode.LIVE_WITH_BACKUPS,
+        streaming_mode: StreamingMode = StreamingMode.LIVE_ONLY,
         retry_policy: RetryPolicyPy | None = None,
         disk_backup_policy: DiskBackupPolicyPy | None = None,
         checkpoint_interval_seconds: int | None = None,
@@ -344,7 +346,8 @@ async def _create(
             run: The run to associate with ingestion. Can be a Run, RunCreate, dict, or run ID string.
             asset_tags: Tags to associate with the asset.
             asset_metadata: Metadata to associate with the asset.
-            streaming_mode: Transport mode for the stream. Defaults to LIVE_WITH_BACKUPS.
+            streaming_mode: Transport mode for the stream. Defaults to LIVE_ONLY, without disk
+                backups. Set to LIVE_WITH_BACKUPS to opt in to local disk backups.
             retry_policy: Retry policy for LIVE_WITH_BACKUPS mode.
             disk_backup_policy: Disk backup policy for LIVE_WITH_BACKUPS or FILE_BACKUP mode.
             checkpoint_interval_seconds: Checkpoint interval in seconds (LIVE_WITH_BACKUPS only).
@@ -686,7 +689,7 @@ async def _create(
         run: RunCreate | dict | str | Run | RunFormPy | None = None,
         asset_tags: list[str] | list[Tag] | None = None,
         asset_metadata: dict[str, str | float | bool] | None = None,
-        streaming_mode: StreamingMode = StreamingMode.LIVE_WITH_BACKUPS,
+        streaming_mode: StreamingMode = StreamingMode.LIVE_ONLY,
         retry_policy: RetryPolicyPy | None = None,
         disk_backup_policy: DiskBackupPolicyPy | None = None,
         checkpoint_interval_seconds: int | None = None,