SAC-31984: Code Refactor

README.md

@@ -13,6 +13,21 @@ It:
     - [Company Call Logs](https://developers.ringcentral.com/api-reference#Call-Log-loadCompanyCallLog)
     - [SMS/MMS/Voicemal/Fax](https://developers.ringcentral.com/api-reference#SMS-and-MMS-listMessages)
 
+### Replication
+
+| Stream | Replication Method | Bookmark |
+| --- | --- | --- |
+| `contacts` | FULL_TABLE | n/a |
+| `company_call_log` | INCREMENTAL | `processedUntil` |
+| `call_log` (User Call Logs) | INCREMENTAL | `processedUntil` |
+| `messages` (SMS/MMS/Voicemail/Fax) | INCREMENTAL | `processedUntil` |
+
+`call_log` and `messages` are extension-based streams: to sync them, the tap first fetches the list of
+extensions from the `contacts` directory (even if `contacts` itself isn't selected), then syncs each
+extension's records individually. If the `contacts` directory can't be read (for example due to a
+permissions error), `call_log` and `messages` are skipped for that run and an error is logged. Streams
+that don't depend on contacts, such as `company_call_log`, are unaffected and continue to sync normally.
+
 ### Quick Start
 
 #### 1. Install
@@ -47,6 +62,10 @@ The following permissions are required:
 - Read Call Log
 - Read Messages
 
+Note: `Read Accounts` is required to list extensions via the `contacts` directory, which `call_log`
+and `messages` depend on. If this permission is missing or the directory otherwise can't be read,
+those two streams will be skipped rather than syncing incomplete data.
+
 #### 3. Create the config file.
 
 There is a template you can use at `config.json.example`, just copy it to `config.json` in the repo root and insert your credentials. You will initially need to use the sandbox `api_url` (eg. `platform.devtest.ringcentral.com`), but after graduating from the dev requirements, you will be able to switch this to use the production API endpoint.

tap_ringcentral/__init__.py

@@ -8,8 +8,10 @@
 
 from tap_ringcentral.discover import discover
 
-from tap_ringcentral.client import RingCentralClient
+import tap_ringcentral.cache
+from tap_ringcentral.client import RingCentralClient, RingCentralForbiddenError
 from tap_ringcentral.streams import AVAILABLE_STREAMS
+from tap_ringcentral.streams.contacts import ContactsStream
 
 LOGGER = singer.get_logger()  # noqa
 
@@ -34,25 +36,90 @@ def do_discover(self):
         json.dump(catalog.to_dict(), sys.stdout, indent=2)
         LOGGER.info("Finished discover")
 
+    def _prefill_contacts_cache(self):
+        """Fill contacts cache before sync if any selected stream requires it.
+
+        Returns False if a required contacts fetch failed, so the caller
+        can skip dependent streams instead of silently syncing zero records.
+        """
+        selected = {s.stream for s in self.catalog.get_selected_streams(self.state)}
+        needs_contacts = any(
+            'contacts' in getattr(self.available_streams.get(name), 'REQUIRES', [])
+            for name in selected
+        )
+
+        # Skip pre-fill if contacts is selected — its own sync will fill the cache first
+        if needs_contacts and 'contacts' not in selected and not tap_ringcentral.cache.contacts:
+            LOGGER.info('Pre-filling contacts cache for extension-based streams')
+            try:
+                ContactsStream(self.config, self.state, None, self.client).fill_cache()
+            except RingCentralForbiddenError as exc:
+                LOGGER.error(
+                    'Could not pre-fill contacts cache: %s. '
+                    'Streams that depend on contacts will be skipped this run.',
+                    str(exc)
+                )
+                return False
+            LOGGER.info('Contacts cache filled with %d extensions', len(tap_ringcentral.cache.contacts))
+
+        return True
 
     # Sync the streams in the order specified in the
     # streams/__init__.py list of AVAILABLE_STREAMS
     def do_sync(self):
         LOGGER.info("Starting sync.")
+        contacts_available = self._prefill_contacts_cache()
+
+        selected = self.catalog.get_selected_streams(self.state)
+
+        # Ensure streams that others depend on sync first regardless of catalog order
+        def _sync_order(entry):
+            """Determine the sync order based on dependencies.
+
+            Streams that are required by others should sync first.
+            Returns a tuple where the first element indicates priority.
+
+            Eg:
+                (0, 'contacts') means this stream has no dependencies and should sync early.
+                (1, 'messages') means this stream has dependencies and should sync later.
+            """
+            cls = self.available_streams.get(entry.stream)
+            requires = getattr(cls, 'REQUIRES', []) if cls else []
+            return (1 if requires else 0, entry.stream)
 
-        for stream_to_sync in self.catalog.get_selected_streams(self.state):
-            stream_obj = self.available_streams[stream_to_sync.stream](
+        selected = sorted(selected, key=_sync_order)
+
+        for stream_to_sync in selected:
+            stream_name = stream_to_sync.stream
+            stream_cls = self.available_streams[stream_name]
+
+            if not contacts_available and 'contacts' in getattr(stream_cls, 'REQUIRES', []):
+                LOGGER.error(
+                    'Skipping stream %s: requires contacts, which failed to load this run.',
+                    stream_name
+                )
+                continue
+
+            # Track currently syncing stream
+            singer.set_currently_syncing(self.state, stream_name)
+            singer.write_state(self.state)
+            LOGGER.info('Currently syncing: %s', stream_name)
+
+            stream_obj = stream_cls(
                         self.config, self.state, stream_to_sync, self.client
                     )
             try:
                 stream_obj.state = self.state
                 stream_obj.sync()
                 self.state = stream_obj.state
+
+                # Clear currently_syncing after successful sync
+                singer.set_currently_syncing(self.state, None)
+                singer.write_state(self.state)
             except Exception as e:
                 LOGGER.error(str(e))
                 LOGGER.error('Failed to sync endpoint {}, moving on!'
                              .format(stream_obj.TABLE))
-                raise e
 
         self.save_state(self.state)
 

tap_ringcentral/schema.py

@@ -35,12 +35,8 @@ def get_schemas():
             replication_method=getattr(stream_metadata, "REPLICATION_METHOD", None),
         )
         mdata = metadata.to_map(mdata)
-        automatic_keys = getattr(stream_metadata, "REPLICATION_KEYS", [])
-        for field_name in schema["properties"].keys():
-            if field_name in automatic_keys:
-                mdata = metadata.write(
-                    mdata, ("properties", field_name), "inclusion", "automatic"
-                )
+        for key in getattr(stream_metadata, "REPLICATION_KEYS", []):
+            mdata = metadata.write(mdata, ("properties", key), "inclusion", "automatic")
         mdata = metadata.to_list(mdata)
         field_metadata[stream_name] = mdata
 

tap_ringcentral/schemas/call_log.json

@@ -14,11 +14,11 @@
             "type": ["null", "string"]
         },
         "startTime": {
-            "type": "string",
+            "type": ["null", "string"],
             "format": "date-time"
         },
         "duration": {
-            "type": "integer"
+            "type": ["null", "integer"]
         },
         "type": {
             "type": ["null", "string"]
@@ -32,12 +32,31 @@
         "result": {
             "type": ["null", "string"]
         },
-        "deleted": {},
-        "reason": {},
-        "reasonDescription": {},
-        "message": {},
+        "deleted": {
+            "type": ["null", "boolean"]
+        },
+        "reason": {
+            "type": ["null", "string"]
+        },
+        "reasonDescription": {
+            "type": ["null", "string"]
+        },
+        "message": {
+            "type": ["null", "object"],
+            "properties": {
+                "id": {
+                    "type": ["null", "string"]
+                },
+                "type": {
+                    "type": ["null", "string"]
+                },
+                "uri": {
+                    "type": ["null", "string"]
+                }
+            }
+        },
         "to": {
-            "type": "object",
+            "type": ["null", "object"],
             "properties": {
                 "name": {
                     "type": ["null", "string"]
@@ -54,7 +73,7 @@
             }
         },
         "from": {
-            "type": "object",
+            "type": ["null", "object"],
             "properties": {
                 "phoneNumber": {
                     "type": ["null", "string"]
@@ -71,15 +90,19 @@
             }
         },
         "extension": {
-            "type": "object",
+            "type": ["null", "object"],
             "properties": {
                 "uri": {
                     "type": ["null", "string"]
                 },
                 "id": {
-                    "type": "integer"
+                    "type": ["null", "integer"]
                 }
             }
+        },
+        "processedUntil": {
+            "type": ["null", "string"],
+            "format": "date-time"
         }
     }
 }

tap_ringcentral/schemas/company_call_log.json

@@ -14,11 +14,11 @@
             "type": ["null", "string"]
         },
         "startTime": {
-            "type": "string",
+            "type": ["null", "string"],
             "format": "date-time"
         },
         "duration": {
-            "type": "integer"
+            "type": ["null", "integer"]
         },
         "type": {
             "type": ["null", "string"]
@@ -32,12 +32,31 @@
         "result": {
             "type": ["null", "string"]
         },
-        "deleted": {},
-        "reason": {},
-        "reasonDescription": {},
-        "message": {},
+        "deleted": {
+            "type": ["null", "boolean"]
+        },
+        "reason": {
+            "type": ["null", "string"]
+        },
+        "reasonDescription": {
+            "type": ["null", "string"]
+        },
+        "message": {
+            "type": ["null", "object"],
+            "properties": {
+                "id": {
+                    "type": ["null", "string"]
+                },
+                "type": {
+                    "type": ["null", "string"]
+                },
+                "uri": {
+                    "type": ["null", "string"]
+                }
+            }
+        },
         "to": {
-            "type": "object",
+            "type": ["null", "object"],
             "properties": {
                 "name": {
                     "type": ["null", "string"]
@@ -51,7 +70,7 @@
             }
         },
         "from": {
-            "type": "object",
+            "type": ["null", "object"],
             "properties": {
                 "phoneNumber": {
                     "type": ["null", "string"]
@@ -65,15 +84,19 @@
             }
         },
         "extension": {
-            "type": "object",
+            "type": ["null", "object"],
             "properties": {
                 "uri": {
                     "type": ["null", "string"]
                 },
                 "id": {
-                    "type": "integer"
+                    "type": ["null", "integer"]
                 }
             }
+        },
+        "processedUntil": {
+            "type": ["null", "string"],
+            "format": "date-time"
         }
     }
 }

tap_ringcentral/schemas/contacts.json

@@ -2,53 +2,53 @@
     "type": "object",
     "properties": {
         "id": {
-            "type": "integer"
+            "type": ["null", "integer"]
         },
         "type": {
-            "type": "string"
+            "type": ["null", "string"]
         },
         "status": {
-            "type": "string"
+            "type": ["null", "string"]
         },
         "firstName": {
-            "type": "string"
+            "type": ["null", "string"]
         },
         "lastName": {
-            "type": "string"
+            "type": ["null", "string"]
         },
         "email": {
-            "type": "string"
+            "type": ["null", "string"]
         },
         "extensionNumber": {
-            "type": "integer"
+            "type": ["null", "integer"]
         },
         "name": {
-            "type": "string"
+            "type": ["null", "string"]
         },
         "jobTitle": {
-            "type": "string"
+            "type": ["null", "string"]
         },
         "account": {
-            "type": "object",
+            "type": ["null", "object"],
             "properties": {
                 "id": {
-                    "type": "integer"
+                    "type": ["null", "integer"]
                 }
             }
         },
         "phoneNumbers": {
-            "type": "array",
+            "type": ["null", "array"],
             "items": {
-                "type": "object",
+                "type": ["null", "object"],
                 "properties": {
                     "type": {
-                        "type": "string"
+                        "type": ["null", "string"]
                     },
                     "phoneNumber": {
-                        "type": "string"
+                        "type": ["null", "string"]
                     },
                     "usageType": {
-                        "type": "string"
+                        "type": ["null", "string"]
                     }
                 }
             }