Catalog: Attach namespace to stream

airbyte-cdk/python/airbyte_cdk/sources/streams/core.py

@@ -111,7 +111,10 @@ def get_json_schema(self) -> Mapping[str, Any]:
 
     def as_airbyte_stream(self) -> AirbyteStream:
         stream = AirbyteStream(name=self.name, json_schema=dict(self.get_json_schema()), supported_sync_modes=[SyncMode.full_refresh])
-
+
+        if self.namespace and self.namespace != "":
+            stream.namespace = self.namespace
+
         if self.supports_incremental:
             stream.source_defined_cursor = self.source_defined_cursor
             stream.supported_sync_modes.append(SyncMode.incremental)  # type: ignore
@@ -141,6 +144,14 @@ def cursor_field(self) -> Union[str, List[str]]:
         """
         return []
 
+    @property
+    def namespace(self) -> Optional[str]:
+        """
+        Override to return the namespace of this stream, e.g. the Postgres schema which this stream will emit records for.
+        :return: A string containing the name of the namespace.
+        """
+        return None
+
     @property
     def source_defined_cursor(self) -> bool:
         """

airbyte-cdk/python/unit_tests/sources/streams/test_streams_core.py

@@ -99,6 +99,35 @@ def test_supports_incremental_cursor_not_set():
     assert not test_stream.supports_incremental
 
 
+def test_namespace_set():
+    """
+    Should allow namespace property to be set.
+    """
+    test_stream = StreamStubFullRefresh()
+    test_stream.namespace = "test_namespace"
+
+    assert test_stream.namespace == "test_namespace"
+
+
+def test_namespace_set_to_empty_string():
+    """
+    Should not set namespace property if equal to empty string.
+    """
+    test_stream = StreamStubFullRefresh()
+    test_stream.namespace = ""
+
+    assert test_stream.namespace == None    
+
+
+def test_namespace_not_set():
+    """
+    Should be equal to unset value of None.
+    """
+    test_stream = StreamStubFullRefresh()
+
+    assert test_stream.namespace == None
+
+
 @pytest.mark.parametrize(
     "test_input, expected",
     [("key", [["key"]]), (["key1", "key2"], [["key1"], ["key2"]]), ([["key1", "key2"], ["key3"]], [["key1", "key2"], ["key3"]])],