rely on config for name source

Author: jcatt-sf

CHANGELOG.md

@@ -8,11 +8,11 @@
 
   New methods let an entry point process a Data Lake Object's Change Data Feed continuously instead of reading a bounded snapshot:
 
-  - `read_dlo_deltas(name)` / `read_dmo_deltas(name)` – return a streaming DataFrame over the object's change feed.
+  - `read_dlo_deltas()` / `read_dmo_deltas()` – return a streaming DataFrame over the object's change feed.
   - `write_dlo_deltas(name, dataframe)` – start a streaming query that writes each micro-batch to the target DLO and return the `StreamingQuery` handle.
 
   ```python
-  deltas = client.read_dlo_deltas("Input__dll")
+  deltas = client.read_dlo_deltas()
   transformed = deltas.withColumn("description__c", upper(col("description__c")))
   query = client.write_dlo_deltas("Output__dll", transformed)
   query.awaitTermination()

README.md

@@ -155,8 +155,8 @@ You should only need the following methods:
 * `write_to_dmo(name, spark_dataframe, write_mode)` – Write to a Data Lake Object by name with a Spark dataframe
 
 For streaming (delta) transforms, the streaming counterparts are:
-* `read_dlo_deltas(name)` – Read the streaming change feed (deltas) of a Data Lake Object as a streaming DataFrame
-* `read_dmo_deltas(name)` – Read the streaming change feed (deltas) of a Data Model Object as a streaming DataFrame
+* `read_dlo_deltas()` – Read the streaming change feed (deltas) of a Data Lake Object as a streaming DataFrame.
+* `read_dmo_deltas()` – Read the streaming change feed (deltas) of a Data Model Object as a streaming DataFrame.
 * `write_dlo_deltas(name, spark_dataframe)` – Write a streaming DataFrame of deltas to a Data Lake Object; returns the started `StreamingQuery`
 
 For example:
@@ -186,7 +186,8 @@ from datacustomcode import Client
 client = Client()
 
 # read_dlo_deltas returns a *streaming* DataFrame over the change feed.
-deltas = client.read_dlo_deltas("Input__dll")
+# The runtime resolves the single streaming source, so no name is passed.
+deltas = client.read_dlo_deltas()
 
 # Ordinary PySpark transform.
 transformed = deltas.withColumn("description__c", upper(col("description__c")))

src/datacustomcode/client.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 from enum import Enum
+import os
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -45,6 +46,15 @@
     from datacustomcode.spark.base import BaseSparkSessionProvider
 
 
+_STREAMING_SOURCE_ENV = "BYOC_STREAMING_SOURCE_NAME"
+_STREAMING_SOURCE_FALLBACK = "<streaming delta source>"
+
+
+def _streaming_source_name() -> str:
+    """Return the runtime streaming source name, or a readable fallback."""
+    return os.environ.get(_STREAMING_SOURCE_ENV, _STREAMING_SOURCE_FALLBACK)
+
+
 def _build_spark_llm_gateway() -> "SparkLLMGateway":
     """Instantiate the SDK-configured :class:`SparkLLMGateway`.
 
@@ -336,7 +346,7 @@ def read_dmo(self, name: str) -> PySparkDataFrame:
         self._record_dmo_access(name)
         return self._reader.read_dmo(name)  # type: ignore[no-any-return]
 
-    def read_dlo_deltas(self, name: str) -> PySparkDataFrame:
+    def read_dlo_deltas(self) -> PySparkDataFrame:
         """Read the streaming change feed (deltas) for a DLO from Data Cloud.
 
         Streaming counterpart to :meth:`read_dlo`, for use in a streaming
@@ -345,29 +355,24 @@ def read_dlo_deltas(self, name: str) -> PySparkDataFrame:
         ``_commit_*``) alongside the source columns. Pair with
         :meth:`write_dlo_deltas` to write the transformed stream back to a DLO.
 
-        Args:
-            name: The name of the DLO to read deltas from.
-
         Returns:
             A streaming PySpark DataFrame over the DLO change feed.
         """
-        self._record_dlo_access(name)
-        return self._reader.read_dlo_deltas(name)  # type: ignore[no-any-return]
+        self._record_dlo_access(_streaming_source_name())
+        return self._reader.read_dlo_deltas()  # type: ignore[no-any-return]
 
-    def read_dmo_deltas(self, name: str) -> PySparkDataFrame:
+    def read_dmo_deltas(self) -> PySparkDataFrame:
         """Read the streaming change feed (deltas) for a DMO from Data Cloud.
 
         Streaming counterpart to :meth:`read_dmo`. See :meth:`read_dlo_deltas`
-        for the shape of the returned change feed.
-
-        Args:
-            name: The name of the DMO to read deltas from.
+        for the shape of the returned change feed and why no source name is
+        passed.
 
         Returns:
             A streaming PySpark DataFrame over the DMO change feed.
         """
-        self._record_dmo_access(name)
-        return self._reader.read_dmo_deltas(name)  # type: ignore[no-any-return]
+        self._record_dmo_access(_streaming_source_name())
+        return self._reader.read_dmo_deltas()  # type: ignore[no-any-return]
 
     def write_to_dlo(
         self, name: str, dataframe: PySparkDataFrame, write_mode: WriteMode, **kwargs

src/datacustomcode/io/reader/base.py

@@ -42,11 +42,7 @@ def read_dmo(
         schema: Union[AtomicType, StructType, str, None] = None,
     ) -> PySparkDataFrame: ...
 
-    def read_dlo_deltas(
-        self,
-        name: str,
-        schema: Union[AtomicType, StructType, str, None] = None,
-    ) -> PySparkDataFrame:
+    def read_dlo_deltas(self) -> PySparkDataFrame:
         """Read the streaming change feed (deltas) for a Data Lake Object.
 
         This is the streaming counterpart to :meth:`read_dlo`. It returns a
@@ -56,11 +52,6 @@ def read_dlo_deltas(
         base implementation raises :class:`NotImplementedError` so local
         readers that do not support streaming fail clearly.
 
-        Args:
-            name: Data Lake Object name.
-            schema: Accepted for parity with :meth:`read_dlo`; implementations
-                may ignore it.
-
         Returns:
             A streaming PySpark DataFrame over the DLO change feed.
 
@@ -74,21 +65,12 @@ def read_dlo_deltas(
             "deltas."
         )
 
-    def read_dmo_deltas(
-        self,
-        name: str,
-        schema: Union[AtomicType, StructType, str, None] = None,
-    ) -> PySparkDataFrame:
+    def read_dmo_deltas(self) -> PySparkDataFrame:
         """Read the streaming change feed (deltas) for a Data Model Object.
 
         Streaming counterpart to :meth:`read_dmo`. See :meth:`read_dlo_deltas`
         for behavior and the local-development caveat.
 
-        Args:
-            name: Data Model Object name.
-            schema: Accepted for parity with :meth:`read_dmo`; implementations
-                may ignore it.
-
         Returns:
             A streaming PySpark DataFrame over the DMO change feed.
 

src/datacustomcode/templates/script/examples/streaming_deltas/entrypoint.py

@@ -4,7 +4,7 @@
 of ``read_dlo`` / ``write_to_dlo`` (which read and write a bounded snapshot), it
 uses the streaming delta methods:
 
-* ``client.read_dlo_deltas(name)`` returns a *streaming* DataFrame over the
+* ``client.read_dlo_deltas()`` returns a *streaming* DataFrame over the
   Change Data Feed of the source DLO. Each row carries the source columns plus
   change-feed metadata columns (``_record_type``, ``_commit_*``).
 * ``client.write_dlo_deltas(name, df)`` starts a streaming query that writes
@@ -30,7 +30,7 @@ def main():
     client = Client()
 
     # Streaming DataFrame over the source DLO's change feed.
-    deltas = client.read_dlo_deltas("Account_std__dll")
+    deltas = client.read_dlo_deltas()
 
     # Ordinary PySpark transform.
     transformed = deltas.withColumn("description__c", upper(col("description__c")))

tests/io/reader/test_query_api.py

@@ -227,13 +227,13 @@ def test_read_dlo(
     def test_read_dlo_deltas_not_supported_locally(self, reader_without_init):
         """Streaming delta reads are not supported by the local reader."""
         with pytest.raises(NotImplementedError) as exc_info:
-            reader_without_init.read_dlo_deltas("test_dlo")
+            reader_without_init.read_dlo_deltas()
         assert "read_dlo_deltas" in str(exc_info.value)
 
     def test_read_dmo_deltas_not_supported_locally(self, reader_without_init):
         """Streaming delta reads are not supported by the local reader."""
         with pytest.raises(NotImplementedError) as exc_info:
-            reader_without_init.read_dmo_deltas("test_dmo")
+            reader_without_init.read_dmo_deltas()
         assert "read_dmo_deltas" in str(exc_info.value)
 
     def test_read_dlo_with_schema(

tests/test_client.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from unittest.mock import MagicMock, patch
 
 from pyspark.sql import DataFrame, SparkSession
@@ -200,11 +201,32 @@ def test_read_dlo_deltas(self, reset_client, mock_spark):
         reader.read_dlo_deltas.return_value = mock_df
 
         client = Client(reader=reader, writer=writer)
-        result = client.read_dlo_deltas("test_dlo")
+        with patch.dict("os.environ", {}, clear=False):
+            os.environ.pop("BYOC_STREAMING_SOURCE_NAME", None)
+            result = client.read_dlo_deltas()
 
-        reader.read_dlo_deltas.assert_called_once_with("test_dlo")
+        reader.read_dlo_deltas.assert_called_once_with()
         assert result is mock_df
-        assert "test_dlo" in client._data_layer_history[DataCloudObjectType.DLO]
+        assert (
+            "<streaming delta source>"
+            in client._data_layer_history[DataCloudObjectType.DLO]
+        )
+
+    def test_read_dlo_deltas_records_runtime_source_name(
+        self, reset_client, mock_spark
+    ):
+        """The runtime source env var populates the access-history entry."""
+        reader = MagicMock(spec=BaseDataCloudReader)
+        writer = MagicMock(spec=BaseDataCloudWriter)
+        reader.read_dlo_deltas.return_value = MagicMock(spec=DataFrame)
+
+        client = Client(reader=reader, writer=writer)
+        with patch.dict(
+            "os.environ", {"BYOC_STREAMING_SOURCE_NAME": "Account_std__dll"}
+        ):
+            client.read_dlo_deltas()
+
+        assert "Account_std__dll" in client._data_layer_history[DataCloudObjectType.DLO]
 
     def test_read_dmo_deltas(self, reset_client, mock_spark):
         reader = MagicMock(spec=BaseDataCloudReader)
@@ -213,11 +235,16 @@ def test_read_dmo_deltas(self, reset_client, mock_spark):
         reader.read_dmo_deltas.return_value = mock_df
 
         client = Client(reader=reader, writer=writer)
-        result = client.read_dmo_deltas("test_dmo")
+        with patch.dict(
+            "os.environ", {"BYOC_STREAMING_SOURCE_NAME": "Account_model__dlm"}
+        ):
+            result = client.read_dmo_deltas()
 
-        reader.read_dmo_deltas.assert_called_once_with("test_dmo")
+        reader.read_dmo_deltas.assert_called_once_with()
         assert result is mock_df
-        assert "test_dmo" in client._data_layer_history[DataCloudObjectType.DMO]
+        assert (
+            "Account_model__dlm" in client._data_layer_history[DataCloudObjectType.DMO]
+        )
 
     def test_write_dlo_deltas(self, reset_client, mock_spark):
         reader = MagicMock(spec=BaseDataCloudReader)
@@ -262,10 +289,11 @@ def test_streaming_read_write_flow(self, reset_client, mock_spark):
 
         client = Client(reader=reader, writer=writer)
 
-        df = client.read_dlo_deltas("source_dll")
+        with patch.dict("os.environ", {"BYOC_STREAMING_SOURCE_NAME": "source_dll"}):
+            df = client.read_dlo_deltas()
         client.write_dlo_deltas("target_dll", df)
 
-        reader.read_dlo_deltas.assert_called_once_with("source_dll")
+        reader.read_dlo_deltas.assert_called_once_with()
         writer.write_dlo_deltas.assert_called_once_with("target_dll", stream_df)
         assert "source_dll" in client._data_layer_history[DataCloudObjectType.DLO]