Add issued invoices resource to the Airbyte connector

airbyte-integrations/connectors/source-orb/integration_tests/configured_catalog.json

@@ -36,6 +36,18 @@
       "sync_mode": "incremental",
       "destination_sync_mode": "overwrite"
     },
+    {
+      "stream": {
+        "name": "invoices",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["invoice_date"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite"
+    },
     {
       "stream": {
         "name": "credits_ledger_entries",

airbyte-integrations/connectors/source-orb/source_orb/schemas/invoices.json

@@ -0,0 +1,38 @@
+{
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": ["null", "object"],
+    "properties": {
+      "id": {
+        "type": "string"
+      },
+      "created_at": {
+        "type": ["null", "string"],
+        "format": "date-time"
+      },
+      "invoice_date": {
+        "type": ["string"],
+        "format": "date-time"
+      },
+      "due_date": {
+        "type": ["string"],
+        "format": "date-time"
+      },
+      "invoice_pdf": {
+        "type": ["null", "string"]
+      },
+      "subtotal": {
+        "type": ["string"]
+      },
+      "total": {
+        "type": ["string"]
+      },
+      "amount_due": {
+        "type": ["string"]
+      },
+      "status": {
+        "type": ["string"]
+      }
+    },
+    "required": ["id", "created_at", "invoice_date", "due_date", "subtotal", "total", "amount_due", "status"]
+  }
+

airbyte-integrations/connectors/source-orb/source_orb/source.py

@@ -142,7 +142,7 @@ def request_params(self, stream_state: Mapping[str, Any], **kwargs) -> MutableMa
         if state_based_start_timestamp:
             # This may (reasonably) override the existing `created_at[gte]` set based on the start_date
             # of the stream, as configured.
-            params["created_at[gte]"] = state_based_start_timestamp
+            params[f"{self.cursor_field}[gte]"] = state_based_start_timestamp
         return params
 
 
@@ -447,6 +447,29 @@ class Plans(IncrementalOrbStream):
     def path(self, **kwargs) -> str:
         return "plans"
 
+class Invoices(IncrementalOrbStream):
+    """
+    Fetches issued invoices.
+    API Docs: https://docs.withorb.com/docs/orb-docs/api-reference/operations/list-invoices
+    """
+
+    @property
+    def cursor_field(self) -> str:
+        """
+        Invoices created in the past may be newly issued, so we store state on 
+        `invoice_date` instead.
+        """
+        return "invoice_date"
+
+    def path(self, **kwargs) -> str:
+        return "invoices"
+
+    def request_params(self, stream_state: Mapping[str, Any], **kwargs) -> MutableMapping[str, Any]:
+        request_params = super().request_params(stream_state, **kwargs)
+        # This doesn't make sense for invoices, since the cursor field is `invoice_date`, and the
+        # created_at for an invoice isn't meaningful, so any attempt to use it should be prevented.
+        del request_params["created_at[gte]"]
+        return request_params
 
 class CreditsLedgerEntries(IncrementalOrbStream):
     page_size = 500
@@ -705,6 +728,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             Customers(authenticator=authenticator, lookback_window_days=lookback_window, start_date=start_date),
             Subscriptions(authenticator=authenticator, lookback_window_days=lookback_window, start_date=start_date),
             Plans(authenticator=authenticator, lookback_window_days=lookback_window, start_date=start_date),
+            Invoices(authenticator=authenticator, lookback_window_days=lookback_window, start_date=start_date),
             CreditsLedgerEntries(
                 authenticator=authenticator,
                 lookback_window_days=lookback_window,