fix articles stream (#70)

* fix articles stream

we have to get articles through `get articles` because then we have the statistics object. adding a child stream with the statistics object as `articles_extended` as some users might not be interested in it and it requires a request per article.

* fix linting

* Apply suggestion from @Copilot

Co-authored-by: Copilot <[email protected]>

* Apply suggestion from @Copilot

Co-authored-by: Copilot <[email protected]>

---------

Co-authored-by: Copilot <[email protected]>

Author: tobiascadee

tap_intercom/schemas.py

@@ -4,6 +4,7 @@
     ArrayType,
     BooleanType,
     IntegerType,
+    NumberType,
     ObjectType,
     PropertiesList,
     Property,
@@ -521,6 +522,46 @@
 ).to_dict()
 
 articles_schema = PropertiesList(
+    Property("id", StringType),
+    Property("parent_type", StringType),
+    Property("parent_ids", ArrayType(IntegerType)),
+    Property(
+        "translated_content",
+        ObjectType(*translated_content),
+    ),
+    Property("languages", ArrayType(StringType)),
+    Property(
+        "tags",
+        ObjectType(
+            Property(
+                "tags",
+                ArrayType(
+                    ObjectType(
+                        Property("id", StringType),
+                        Property("name", StringType),
+                        Property("applied_at", IntegerType),
+                        Property(
+                            "applied_by",
+                            ObjectType(
+                                Property("type", StringType),
+                                Property("id", StringType),
+                            ),
+                        ),
+                    ),
+                ),
+            ),
+        ),
+    ),
+    Property("title", StringType),
+    Property("description", StringType),
+    Property("body", StringType),
+    Property("author_id", IntegerType),
+    Property("state", StringType),
+    Property("created_at", IntegerType),
+    Property("updated_at", IntegerType),
+).to_dict()
+
+articles_extended_schema = PropertiesList(
     Property("id", StringType),
     Property("parent_type", StringType),
     Property("parent_ids", ArrayType(IntegerType)),
@@ -562,9 +603,9 @@
             Property("views", IntegerType),
             Property("conversions", IntegerType),
             Property("reactions", IntegerType),
-            Property("happy_reaction_percentage", IntegerType),
-            Property("neutral_reaction_percentage", IntegerType),
-            Property("sad_reaction_percentage", IntegerType),
+            Property("happy_reaction_percentage", NumberType),
+            Property("neutral_reaction_percentage", NumberType),
+            Property("sad_reaction_percentage", NumberType),
         ),
     ),
     Property("created_at", IntegerType),

tap_intercom/streams.py

@@ -2,12 +2,14 @@
 
 from __future__ import annotations
 
+import decimal
 import typing as t
 from urllib.parse import parse_qsl
 
 from tap_intercom.client import IntercomHATEOASPaginator, IntercomStream
 from tap_intercom.schemas import (
     admins_schema,
+    articles_extended_schema,
     articles_schema,
     contacts_schema,
     conversation_parts_schema,
@@ -16,6 +18,9 @@
     teams_schema,
 )
 
+if t.TYPE_CHECKING:
+    import requests
+
 
 class ConversationsStream(IntercomStream):
     """Stream for Intercom conversations."""
@@ -98,8 +103,8 @@ class ArticlesStream(IntercomStream):
     """Stream for Intercom articles."""
 
     name = "articles"
-    path = "/articles/search"
-    records_jsonpath = "$.data.articles[*]"
+    path = "/articles"
+    records_jsonpath = "$.data[*]"
     schema = articles_schema
 
     def get_new_paginator(self) -> IntercomHATEOASPaginator:
@@ -124,11 +129,34 @@ def get_url_params(
         Returns:
             Dictionary of URL parameters.
         """
-        params = {}
+        params = super().get_url_params(context, next_page_token)
         if next_page_token:
             # parse URL for next page
             params.update(dict(parse_qsl(next_page_token.query)))
-            return params
 
-        # default to parent class for initial request
-        return super().get_url_params(context, next_page_token)
+        return params
+
+    def get_child_context(self, record: dict, context: dict | None) -> dict:  # noqa: ARG002
+        """Return a context dictionary for child streams."""
+        return {"article_id": record["id"]}
+
+
+class ArticlesExtendedStream(IntercomStream):
+    """Stream providing extended article data, including statistics, for individual Intercom articles."""
+
+    name = "articles_extended"
+    path = "/articles/{article_id}"
+    records_jsonpath = "$"
+    schema = articles_extended_schema
+    parent_stream_type = ArticlesStream
+
+    def parse_response(self, response: requests.Response) -> t.Iterable[dict]:
+        """Parse the response and return an iterator of result records.
+
+        Args:
+            response: A raw :class:`requests.Response`
+
+        Yields:
+            One item for every item found in the response.
+        """
+        yield response.json(parse_float=decimal.Decimal)

tap_intercom/tap.py

@@ -58,6 +58,7 @@ def discover_streams(self) -> list[streams.IntercomStream]:
             streams.TeamsStream(self),
             streams.ContactsStream(self),
             streams.ArticlesStream(self),
+            streams.ArticlesExtendedStream(self),
         ]