Adding stream url path link for dynamic schema (#7)

* Added module-path in catalog

* Removed module-path, added path in sync

Author: atttiwari

tap_zoho_crm/discover.py

@@ -31,10 +31,10 @@ def discover(client: Client) -> Catalog:
             raise err
 
         key_properties = metadata.to_map(mdata).get((), {}).get("table-key-properties")
-
+        stream_name = stream_name.lower()
         catalog.streams.append(
             CatalogEntry(
-                stream=stream_name.lower(),
+                stream=stream_name,
                 tap_stream_id=stream_name,
                 key_properties=key_properties,
                 schema=schema,

tap_zoho_crm/sync.py

@@ -4,6 +4,7 @@
 from tap_zoho_crm.streams import STREAMS, abstracts
 from tap_zoho_crm.client import Client
 from tap_zoho_crm.streams.abstracts import IncrementalStream, FullTableStream
+from tap_zoho_crm.schema import get_dynamic_schema
 
 LOGGER = singer.get_logger()
 
@@ -34,7 +35,11 @@ def write_schema(stream, client, streams_to_sync, catalog) -> None:
             stream.child_to_sync.append(child_obj)
 
 
-def build_dynamic_stream(client, catalog_entry: singer.CatalogEntry) -> object:
+def build_dynamic_stream(
+        client,
+        catalog_entry: singer.CatalogEntry,
+        module_path: str
+    ) -> object:
     """Create a dynamic stream instance based on stream_catalog."""
     catalog_metadata = metadata.to_map(catalog_entry.metadata)
 
@@ -49,7 +54,7 @@ def build_dynamic_stream(client, catalog_entry: singer.CatalogEntry) -> object:
         "key_properties": property(lambda self: key_properties),
         "replication_method": property(lambda self: replication_method),
         "replication_keys": property(lambda self: replication_keys),
-        "path": tap_stream_id,
+        "path": module_path,
         "data_key": "data",
         "is_dynamic": True
     }
@@ -86,7 +91,8 @@ def sync(client: Client, config: Dict, catalog: singer.Catalog, state) -> None:
     """
     Sync selected streams from catalog
     """
-
+    dynamic_schemas, _ = get_dynamic_schema(client)
+    dynamic_schema_path = {item.lower(): item for item in dynamic_schemas.keys()}
     streams_to_sync = []
     for stream in catalog.get_selected_streams(state):
         catalog_entry = catalog.get_stream(stream.tap_stream_id)
@@ -104,7 +110,11 @@ def sync(client: Client, config: Dict, catalog: singer.Catalog, state) -> None:
             if stream_name in STREAMS:
                 stream = STREAMS[stream_name](client, catalog.get_stream(stream_name))
             else:
-                stream = build_dynamic_stream(client, catalog.get_stream(stream_name))
+                stream = build_dynamic_stream(
+                    client,
+                    catalog.get_stream(stream_name),
+                    dynamic_schema_path.get(stream_name)
+                )
 
             parent_name = getattr(stream, "parent", None)
             if parent_name:

tests/unittests/test_sync.py

@@ -86,26 +86,26 @@ def test_build_dynamic_stream_full_table(self):
         """Test build_dynamic_stream returns FullTableStream subclass when method is FULL_TABLE."""
         catalog_entry = MagicMock()
         catalog_entry.stream = "accounts"
-        catalog_entry.tap_stream_id = "Accounts"
+        catalog_entry.tap_stream_id = "accounts"
         catalog_entry.key_properties = ["id"]
 
         # Here's the mocked metadata (as a raw dict)
         catalog_entry.metadata = [
             {
                 "breadcrumb": [],
                 "metadata": {
-                    "tap_stream_id": "Accounts",
+                    "tap_stream_id": "accounts",
                     "forced-replication-method": "FULL_TABLE",
                     "valid-replication-keys": ["id"]
                 }
             }
         ]
 
         mock_client = MagicMock()
-        stream_instance = build_dynamic_stream(mock_client, catalog_entry)
+        stream_instance = build_dynamic_stream(mock_client, catalog_entry, "Accounts")
 
         self.assertIsInstance(stream_instance, FullTableStream)
-        self.assertEqual(stream_instance.tap_stream_id, "Accounts")
+        self.assertEqual(stream_instance.tap_stream_id, "accounts")
         self.assertEqual(stream_instance.key_properties, ["id"])
         self.assertEqual(stream_instance.replication_method, "FULL_TABLE")
         self.assertEqual(stream_instance.replication_keys, ["id"])
@@ -117,24 +117,24 @@ def test_build_dynamic_stream_incremental(self):
         """Test build_dynamic_stream returns IncrementalStream subclass when method is INCREMENTAL."""
         catalog_entry = MagicMock()
         catalog_entry.stream = "contacts"
-        catalog_entry.tap_stream_id = "Contacts"
+        catalog_entry.tap_stream_id = "contacts"
         catalog_entry.key_properties = ["id"]
         catalog_entry.metadata = [
             {
                 "breadcrumb": [],
                 "metadata": {
-                    "tap_stream_id": "Contacts",
+                    "tap_stream_id": "contacts",
                     "forced-replication-method": "INCREMENTAL",
                     "valid-replication-keys": ["updated_at"]
                 }
             }
         ]
 
         mock_client = MagicMock()
-        stream_instance = build_dynamic_stream(mock_client, catalog_entry)
+        stream_instance = build_dynamic_stream(mock_client, catalog_entry, "Contacts")
 
         self.assertIsInstance(stream_instance, IncrementalStream)
-        self.assertEqual(stream_instance.tap_stream_id, "Contacts")
+        self.assertEqual(stream_instance.tap_stream_id, "contacts")
         self.assertEqual(stream_instance.replication_method, "INCREMENTAL")
         self.assertEqual(stream_instance.replication_keys, ["updated_at"])
         self.assertEqual(stream_instance.path, "Contacts")