✨ Source Zendesk Support: migration posts child streams to low-code

airbyte-integrations/connectors/source-zendesk-support/source_zendesk_support/manifest.yaml

@@ -353,6 +353,9 @@ definitions:
     $ref: "#/definitions/base_incremental_stream"
     name: "posts"
     primary_key: "id"
+    schema_loader:
+      type: JsonFileSchemaLoader
+      file_path: "./source_zendesk_support/schemas/posts.json"
     incremental_sync:
       $ref: "#/definitions/cursor_incremental_sync"
       start_time_option:
@@ -371,10 +374,118 @@ definitions:
           $ref: "#/definitions/links_next_paginator/pagination_strategy"
           stop_condition: "{{ config.get('ignore_pagination', False) or not response['meta']['has_more'] }}"
       record_selector:
-        $ref: "#/definitions/retriever/record_selector"
+        extractor:
+          type: DpathExtractor
+          field_path: ["posts"]
         schema_normalization: None
-    $parameters:
-      name: "posts"
+
+  post_comments_stream:
+    $ref: "#/definitions/base_incremental_stream"
+    name: "post_comments"
+    primary_key: "id"
+    schema_loader:
+      type: JsonFileSchemaLoader
+      file_path: "./source_zendesk_support/schemas/post_comments.json"
+    incremental_sync:
+      $ref: "#/definitions/cursor_incremental_sync"
+      start_time_option:
+        $ref: "#/definitions/cursor_incremental_sync/start_time_option"
+        field_name: "start_time"
+    retriever:
+      $ref: "#/definitions/retriever"
+      ignore_stream_slicer_parameters_on_paginated_requests: true
+      requester:
+        $ref: "#/definitions/retriever/requester"
+        path: "community/posts/{{ stream_partition.post_id }}/comments"
+      paginator:
+        $ref: "#/definitions/links_next_paginator"
+      partition_router:
+        type: SubstreamPartitionRouter
+        parent_stream_configs:
+          - type: ParentStreamConfig
+            parent_key: "id"
+            partition_field: "post_id"
+            stream:
+              $ref: "#/definitions/posts_stream"
+            incremental_dependency: true
+      record_selector:
+        extractor:
+          type: DpathExtractor
+          field_path: ["comments"]
+    transformations:
+      - type: AddFields
+        fields:
+          - path:
+              - _airbyte_parent_id # used for post_comment_votes stream
+            value: "{{ { 'post_id': record['post_id'], 'comment_id': record['id'] } }}"
+
+  post_votes_stream:
+    $ref: "#/definitions/base_incremental_stream"
+    name: "post_votes"
+    primary_key: "id"
+    schema_loader:
+      type: JsonFileSchemaLoader
+      file_path: "./source_zendesk_support/schemas/votes.json"
+    incremental_sync:
+      $ref: "#/definitions/cursor_incremental_sync"
+      start_time_option:
+        $ref: "#/definitions/cursor_incremental_sync/start_time_option"
+        field_name: "start_time"
+    retriever:
+      $ref: "#/definitions/retriever"
+      ignore_stream_slicer_parameters_on_paginated_requests: true
+      requester:
+        $ref: "#/definitions/retriever/requester"
+        path: "community/posts/{{ stream_partition.post_id }}/votes"
+      paginator:
+        $ref: "#/definitions/links_next_paginator"
+      partition_router:
+        type: SubstreamPartitionRouter
+        parent_stream_configs:
+          - type: ParentStreamConfig
+            parent_key: "id"
+            partition_field: "post_id"
+            stream:
+              $ref: "#/definitions/posts_stream"
+            incremental_dependency: true
+      record_selector:
+        extractor:
+          type: DpathExtractor
+          field_path: ["votes"]
+
+  post_comment_votes_stream:
+    $ref: "#/definitions/base_incremental_stream"
+    name: "post_comment_votes"
+    primary_key: "id"
+    schema_loader:
+      type: JsonFileSchemaLoader
+      file_path: "./source_zendesk_support/schemas/votes.json"
+    incremental_sync:
+      $ref: "#/definitions/cursor_incremental_sync"
+      start_time_option:
+        $ref: "#/definitions/cursor_incremental_sync/start_time_option"
+        field_name: "start_time"
+    retriever:
+      $ref: "#/definitions/retriever"
+      ignore_stream_slicer_parameters_on_paginated_requests: true
+      requester:
+        $ref: "#/definitions/retriever/requester"
+        path: "community/posts/{{ stream_partition.id.post_id }}/comments/{{ stream_partition.id.comment_id }}/votes"
+      paginator:
+        $ref: "#/definitions/links_next_paginator"
+      partition_router:
+        type: SubstreamPartitionRouter
+        parent_stream_configs:
+          - type: ParentStreamConfig
+            parent_key: "_airbyte_parent_id"
+            partition_field: "id"
+            stream:
+              $ref: "#/definitions/post_comments_stream"
+            incremental_dependency: true
+      record_selector:
+        extractor:
+          type: DpathExtractor
+          field_path: ["votes"]
 
   satisfaction_ratings_stream:
     $ref: "#/definitions/base_incremental_stream"
@@ -649,6 +760,9 @@ streams:
   - $ref: "#/definitions/organization_memberships_stream"
   - $ref: "#/definitions/organizations_stream"
   - $ref: "#/definitions/posts_stream"
+  - $ref: "#/definitions/post_comment_votes_stream"
+  - $ref: "#/definitions/post_comments_stream"
+  - $ref: "#/definitions/post_votes_stream"
   - $ref: "#/definitions/satisfaction_ratings_stream"
   - $ref: "#/definitions/ticket_activities_stream"
   - $ref: "#/definitions/ticket_audits_stream"

airbyte-integrations/connectors/source-zendesk-support/source_zendesk_support/source.py

@@ -21,10 +21,6 @@
     ArticleCommentVotes,
     Articles,
     ArticleVotes,
-    PostComments,
-    PostCommentVotes,
-    Posts,
-    PostVotes,
     TicketMetrics,
     Tickets,
     UserIdentities,
@@ -122,9 +118,6 @@ def get_nested_streams(self, config: Mapping[str, Any]) -> List[Stream]:
             ArticleComments(**args),
             ArticleCommentVotes(**args),
             ArticleVotes(**args),
-            PostComments(**args),
-            PostCommentVotes(**args),
-            PostVotes(**args),
             Tickets(**args),
             TicketMetrics(**args),
             UserIdentities(**args),

airbyte-integrations/connectors/source-zendesk-support/source_zendesk_support/streams.py

@@ -455,17 +455,6 @@ class Organizations(SourceZendeskIncrementalExportStream):
     next_page_field: str = "next_page"
 
 
-class Posts(CursorPaginationZendeskSupportStream):
-    """Posts stream: https://developer.zendesk.com/api-reference/help_center/help-center-api/posts/#list-posts"""
-
-    use_cache = True
-
-    cursor_field = "updated_at"
-
-    def path(self, **kwargs):
-        return "community/posts"
-
-
 class Tickets(SourceZendeskIncrementalExportStream):
     """Tickets stream: https://developer.zendesk.com/api-reference/ticketing/ticket-management/incremental_exports/#incremental-ticket-export-time-based"""
 
@@ -1055,66 +1044,13 @@ def request_params(
         return req_params
 
 
-class PostComments(CursorPaginationZendeskSupportStream, HttpSubStream):
-    """Post Comments Stream: https://developer.zendesk.com/api-reference/help_center/help-center-api/post_comments/"""
-
-    response_list_name = "comments"
-
-    def __init__(self, **kwargs):
-        parent = Posts(**kwargs)
-        super().__init__(parent=parent, **kwargs)
-
-    def path(
-        self,
-        *,
-        stream_state: Mapping[str, Any] = None,
-        stream_slice: Mapping[str, Any] = None,
-        next_page_token: Mapping[str, Any] = None,
-    ) -> str:
-        post_id = stream_slice.get("parent").get("id")
-        return f"community/posts/{post_id}/comments"
-
-
 class AbstractVotes(CursorPaginationZendeskSupportStream, ABC):
     response_list_name = "votes"
 
     def get_json_schema(self) -> Mapping[str, Any]:
         return ResourceSchemaLoader(package_name_from_class(self.__class__)).get_schema("votes")
 
 
-class PostVotes(AbstractVotes, HttpSubStream):
-    def __init__(self, **kwargs):
-        parent = Posts(**kwargs)
-        super().__init__(parent=parent, **kwargs)
-
-    def path(
-        self,
-        *,
-        stream_state: Mapping[str, Any] = None,
-        stream_slice: Mapping[str, Any] = None,
-        next_page_token: Mapping[str, Any] = None,
-    ) -> str:
-        post_id = stream_slice.get("parent").get("id")
-        return f"community/posts/{post_id}/votes"
-
-
-class PostCommentVotes(AbstractVotes, HttpSubStream):
-    def __init__(self, **kwargs):
-        parent = PostComments(**kwargs)
-        super().__init__(parent=parent, **kwargs)
-
-    def path(
-        self,
-        *,
-        stream_state: Mapping[str, Any] = None,
-        stream_slice: Mapping[str, Any] = None,
-        next_page_token: Mapping[str, Any] = None,
-    ) -> str:
-        post_id = stream_slice.get("parent").get("post_id")
-        comment_id = stream_slice.get("parent").get("id")
-        return f"community/posts/{post_id}/comments/{comment_id}/votes"
-
-
 class Articles(SourceZendeskIncrementalExportStream):
     """Articles Stream: https://developer.zendesk.com/api-reference/help_center/help-center-api/articles/#list-articles"""
 

airbyte-integrations/connectors/source-zendesk-support/unit_tests/integrations/helpers.py

@@ -1,4 +1,5 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+from typing import Optional
 
 import pendulum
 from pendulum.datetime import DateTime
@@ -47,12 +48,14 @@ def given_ticket_forms(
     return ticket_forms_record_builder
 
 
-def given_posts(http_mocker: HttpMocker, start_date: DateTime, api_token_authenticator: ApiTokenAuthenticator) -> PostsRecordBuilder:
+def given_posts(
+    http_mocker: HttpMocker, start_date: DateTime, api_token_authenticator: ApiTokenAuthenticator, updated_at: Optional[DateTime] = None
+) -> PostsRecordBuilder:
     """
     Posts requests setup
     """
     posts_record_builder = PostsRecordBuilder.posts_record().with_field(
-        FieldPath("updated_at"), datetime_to_string(start_date.add(seconds=1))
+        FieldPath("updated_at"), datetime_to_string(updated_at if updated_at else start_date.add(seconds=1))
     )
     http_mocker.get(
         PostsRequestBuilder.posts_endpoint(api_token_authenticator)
@@ -65,13 +68,17 @@ def given_posts(http_mocker: HttpMocker, start_date: DateTime, api_token_authent
 
 
 def given_post_comments(
-    http_mocker: HttpMocker, start_date: DateTime, post_id: int, api_token_authenticator: ApiTokenAuthenticator
+    http_mocker: HttpMocker,
+    start_date: DateTime,
+    post_id: int,
+    api_token_authenticator: ApiTokenAuthenticator,
+    updated_at: Optional[DateTime] = None,
 ) -> PostsCommentsRecordBuilder:
     """
     Post Comments requests setup
     """
-    post_comments_record_builder = PostsCommentsRecordBuilder.posts_commetns_record().with_field(
-        FieldPath("updated_at"), datetime_to_string(start_date.add(seconds=1))
+    post_comments_record_builder = PostsCommentsRecordBuilder.posts_comments_record().with_field(
+        FieldPath("updated_at"), datetime_to_string(updated_at if updated_at else start_date.add(seconds=1))
     )
     http_mocker.get(
         PostsCommentsRequestBuilder.posts_comments_endpoint(api_token_authenticator, post_id)