Connector Facebook-Marketing: update insights streams with custom entries for fields, breakdowns and action_breakdowns

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/source.py

@@ -2,9 +2,12 @@
 # Copyright (c) 2021 Airbyte, Inc., all rights reserved.
 #
 
+import json
 from datetime import datetime
-from typing import Any, List, Mapping, Tuple, Type
+from typing import Any, List, Mapping, Optional, Tuple, Type
 
+from airbyte_cdk.entrypoint import logger
+from airbyte_cdk.logger import AirbyteLogger
 from airbyte_cdk.models import AuthSpecification, ConnectorSpecification, DestinationSyncMode, OAuth2Specification
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.streams import Stream
@@ -25,6 +28,17 @@
 )
 
 
+class InsightConfig(BaseModel):
+
+    name: str = Field(description="The name value of insight")
+
+    fields: Optional[List[str]] = Field(description="A list of chosen fields for fields parameter")
+
+    breakdowns: Optional[List[str]] = Field(description="A list of chosen breakdowns for breakdowns")
+
+    action_breakdowns: Optional[List[str]] = Field(description="A list of chosen action_breakdowns for action_breakdowns")
+
+
 class ConnectorConfig(BaseModel):
     class Config:
         title = "Source Facebook Marketing"
@@ -57,6 +71,12 @@ class Config:
         minimum=1,
         maximum=30,
     )
+    insights: Optional[List[InsightConfig]] = Field(
+        description="A defined list wich contains insights entries, each entry must have a name and can contain these entries(fields, breakdowns or action_breakdowns)",
+        examples=[
+            '[{"name": "AdsInsights","fields": ["account_id","account_name","ad_id","ad_name","adset_id","adset_name","campaign_id","campaign_name","date_start","impressions","spend"],"breakdowns": [],"action_breakdowns": []}]'
+        ],
+    )
 
 
 class SourceFacebookMarketing(AbstractSource):
@@ -94,7 +114,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
             days_per_job=config.insights_days_per_job,
         )
 
-        return [
+        streams = [
             Campaigns(api=api, start_date=config.start_date, include_deleted=config.include_deleted),
             AdSets(api=api, start_date=config.start_date, include_deleted=config.include_deleted),
             Ads(api=api, start_date=config.start_date, include_deleted=config.include_deleted),
@@ -108,6 +128,8 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
             AdsInsightsActionType(**insights_args),
         ]
 
+        return self._update_insights_streams(insights=config.insights, args=insights_args, streams=streams)
+
     def spec(self, *args, **kwargs) -> ConnectorSpecification:
         """
         Returns the spec for this integration. The spec is a JSON-Schema object describing the required configurations (e.g: username and password)
@@ -126,3 +148,28 @@ def spec(self, *args, **kwargs) -> ConnectorSpecification:
                 ),
             ),
         )
+
+    def _update_insights_streams(self, insights, args, streams) -> List[Type[Stream]]:
+        """Update method, if insights have values returns streams replacing the
+        default insights streams else returns streams
+
+        """
+        if not insights:
+            return streams
+
+        insights_custom_streams = list()
+
+        for insight in insights:
+            args["name"] = insight.name
+            args["fields"] = insight.fields
+            args["breakdowns"] = insight.breakdowns
+            args["action_breakdowns"] = insight.action_breakdowns
+            insight_stream = AdsInsights(**args)
+            insights_custom_streams.append(insight_stream)
+
+        new_streams = list()
+        for stream in streams:
+            if stream.name not in [e.name for e in insights_custom_streams]:
+                new_streams.append(stream)
+
+        return new_streams + insights_custom_streams

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/streams.py

@@ -9,6 +9,7 @@
 from datetime import datetime
 from typing import Any, Iterable, Iterator, List, Mapping, MutableMapping, Optional, Sequence
 
+import airbyte_cdk.sources.utils.casing as casing
 import backoff
 import pendulum
 from airbyte_cdk.models import SyncMode
@@ -290,10 +291,32 @@ class AdsInsights(FBMarketingIncrementalStream):
 
     breakdowns = []
 
-    def __init__(self, buffer_days, days_per_job, **kwargs):
+    def __init__(
+        self,
+        buffer_days,
+        days_per_job,
+        name: str = None,
+        fields: List[str] = None,
+        breakdowns: List[str] = None,
+        action_breakdowns: List[str] = None,
+        **kwargs,
+    ):
+
         super().__init__(**kwargs)
         self.lookback_window = pendulum.duration(days=buffer_days)
         self._days_per_job = days_per_job
+        self._fields = fields
+        self.action_breakdowns = action_breakdowns or self.action_breakdowns
+        self.breakdowns = breakdowns or self.breakdowns
+        self._new_class_name = name
+
+    @property
+    def name(self) -> str:
+        """
+        :return: Stream name. By default this is the implementing class name, but it can be overridden as needed.
+        """
+        name = self._new_class_name or self.__class__.__name__
+        return casing.camel_to_snake(name)
 
     def read_records(
         self,
@@ -388,12 +411,16 @@ def get_json_schema(self) -> Mapping[str, Any]:
         :return: A dict of the JSON schema representing this stream.
         """
         schema = ResourceSchemaLoader(package_name_from_class(self.__class__)).get_schema("ads_insights")
+        if self._fields:
+            schema["properties"] = {k: v for k, v in schema["properties"].items() if k in self._fields}
         schema["properties"].update(self._schema_for_breakdowns())
         return schema
 
     @cached_property
     def fields(self) -> List[str]:
         """List of fields that we want to query, for now just all properties from stream's schema"""
+        if self._fields:
+            return self._fields
         schema = ResourceSchemaLoader(package_name_from_class(self.__class__)).get_schema("ads_insights")
         return list(schema.get("properties", {}).keys())
 

docs/integrations/sources/facebook-marketing.md

@@ -89,6 +89,7 @@ See Facebook's [documentation on rate limiting](https://developers.facebook.com/
 
 | Version | Date       | Pull Request | Subject |
 | :------ | :--------  | :-----       | :------ |
+| 0.2.21  | 2021-10-05 | [4864](https://github.com/airbytehq/airbyte/pull/4864) | Update insights streams with custom entries for fields, breakdowns and action_breakdowns |
 | 0.2.20  | 2021-10-04 | [6719](https://github.com/airbytehq/airbyte/pull/6719) | Update version of facebook_bussiness package to 12.0 |
 | 0.2.19  | 2021-09-30 | [6438](https://github.com/airbytehq/airbyte/pull/6438) | Annotate Oauth2 flow initialization parameters in connector specification |
 | 0.2.18  | 2021-09-28 | [6499](https://github.com/airbytehq/airbyte/pull/6499) | Fix field values converting fail |