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

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/schemas/ads_insights_breakdowns.json

@@ -0,0 +1,38 @@
+{
+    "properties": {
+        "action_device": {"type": ["null", "string"]},
+        "action_canvas_component_name": {"type": ["null", "string"]},
+        "action_carousel_card_id": {"type": ["null", "string"]},
+        "action_carousel_card_name": {"type": ["null", "string"]},
+        "action_destination": {"type": ["null", "string"]},
+        "action_reaction": {"type": ["null", "string"]},
+        "action_target_id": {"type": ["null", "string"]},
+        "action_type": {"type": ["null", "string"]},
+        "action_video_sound": {"type": ["null", "string"]},
+        "action_video_type": {"type": ["null", "string"]},
+        "ad_format_asset": {"type": ["null", "string"]},
+        "age": {"type": ["null", "string"]},
+        "app_id": {"type": ["null", "string"]},
+        "body_asset": {"type": ["null", "string"]},
+        "call_to_action_asset": {"type": ["null", "string"]},
+        "country": {"type": ["null", "string"]},
+        "description_asset": {"type": ["null", "string"]},
+        "device_platform": {"type": ["null", "string"]},
+        "dma": {"type": ["null", "string"]},
+        "frequency_value": {"type": ["null", "string"]},
+        "gender": {"type": ["null", "string"]},
+        "hourly_stats_aggregated_by_advertiser_time_zone": {"type": ["null", "string"]},
+        "hourly_stats_aggregated_by_audience_time_zone": {"type": ["null", "string"]},
+        "image_asset": {"type": ["null", "string"]},
+        "impression_device": {"type": ["null", "string"]},
+        "link_url_asset": {"type": ["null", "string"]},
+        "place_page_id": {"type": ["null", "string"]},
+        "platform_position": {"type": ["null", "string"]},
+        "product_id": {"type": ["null", "string"]},
+        "publisher_platform": {"type": ["null", "string"]},
+        "region": {"type": ["null", "string"]},
+        "skan_conversion_id": {"type": ["null", "string"]},
+        "title_asset": {"type": ["null", "string"]},
+        "video_asset": {"type": ["null", "string"]}
+    }
+}

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

@@ -4,13 +4,16 @@
 
 import json
 from datetime import datetime
-from typing import Any, List, Mapping, Optional, Tuple, Type
+from typing import Any, List, Mapping, Optional, Tuple, Type, MutableMapping
+from jsonschema import RefResolver
 
 from airbyte_cdk.entrypoint import logger
 from airbyte_cdk.logger import AirbyteLogger
-from airbyte_cdk.models import AuthSpecification, ConnectorSpecification, DestinationSyncMode, OAuth2Specification
+from airbyte_cdk.models import AirbyteConnectionStatus, AuthSpecification, ConnectorSpecification, DestinationSyncMode, OAuth2Specification, Status
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.streams import Stream
+from airbyte_cdk.sources.streams.core import package_name_from_class
+from airbyte_cdk.sources.utils.schema_helpers import ResourceSchemaLoader
 from pydantic import BaseModel, Field
 from source_facebook_marketing.api import API
 from source_facebook_marketing.streams import (
@@ -32,11 +35,11 @@ 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")
+    fields: Optional[List[str]] = Field(description="A list of chosen fields for fields parameter", default=[])
 
-    breakdowns: Optional[List[str]] = Field(description="A list of chosen breakdowns for breakdowns")
+    breakdowns: Optional[List[str]] = Field(description="A list of chosen breakdowns for breakdowns", default=[])
 
-    action_breakdowns: Optional[List[str]] = Field(description="A list of chosen action_breakdowns for action_breakdowns")
+    action_breakdowns: Optional[List[str]] = Field(description="A list of chosen action_breakdowns for action_breakdowns", default=[])
 
 
 class ConnectorConfig(BaseModel):
@@ -72,10 +75,7 @@ class Config:
         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": []}]'
-        ],
+        description="A list wich contains insights entries, each entry must have a name and can contains fields, breakdowns or action_breakdowns)"
     )
 
 
@@ -130,6 +130,20 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
 
         return self._update_insights_streams(insights=config.insights, args=insights_args, streams=streams)
 
+    def check(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> AirbyteConnectionStatus:
+        """Implements the Check Connection operation from the Airbyte Specification. See https://docs.airbyte.io/architecture/airbyte-specification."""
+        try:
+            check_succeeded, error = self.check_connection(logger, config)
+            if not check_succeeded:
+                return AirbyteConnectionStatus(status=Status.FAILED, message=repr(error))
+        except Exception as e:
+            return AirbyteConnectionStatus(status=Status.FAILED, message=repr(e))
+
+        self._check_insights_entries(config.get('insights', []))
+
+        return AirbyteConnectionStatus(status=Status.SUCCEEDED)
+
+
     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)
@@ -140,12 +154,12 @@ def spec(self, *args, **kwargs) -> ConnectorSpecification:
             changelogUrl="https://docs.airbyte.io/integrations/sources/facebook-marketing",
             supportsIncremental=True,
             supported_destination_sync_modes=[DestinationSyncMode.append],
-            connectionSpecification=ConnectorConfig.schema(),
+            connectionSpecification=expand_local_ref(ConnectorConfig.schema()),
             authSpecification=AuthSpecification(
                 auth_type="oauth2.0",
                 oauth2Specification=OAuth2Specification(
                     rootObject=[], oauthFlowInitParameters=[], oauthFlowOutputParameters=[["access_token"]]
-                ),
+                )
             ),
         )
 
@@ -160,16 +174,53 @@ def _update_insights_streams(self, insights, args, streams) -> List[Type[Stream]
         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
+            args["name"] = f"Custom{insight.name}"
+            args["fields"] = list(set(insight.fields))
+            args["breakdowns"] = list(set(insight.breakdowns))
+            args["action_breakdowns"] = list(set(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 streams + insights_custom_streams
+
+    def _check_insights_entries(self, insights):
 
-        return new_streams + insights_custom_streams
+        default_fields = list(ResourceSchemaLoader(package_name_from_class(self.__class__)).get_schema("ads_insights").get("properties", {}).keys())
+        default_breakdowns = list(ResourceSchemaLoader(package_name_from_class(self.__class__)).get_schema("ads_insights_breakdowns").get("properties", {}).keys())
+        default_actions_breakdowns = [e for e in default_breakdowns if 'action_' in e]
+
+        for insight in insights:
+            if insight.get('fields') and not self._check_values(default_fields, insight.get('fields')):
+                message = f"For {insight.get('name')} that field are not espected in fields"
+                raise Exception("Config validation error: " + message) from None
+            if insight.get('breakdowns') and not self._check_values(default_breakdowns, insight.get('breakdowns')):
+                message = f"For {insight.get('name')} that breakdown are not espected in breakdowns"
+                raise Exception("Config validation error: " + message) from None
+            if insight.get('action_breakdowns') and not self._check_values(default_actions_breakdowns, insight.get('action_breakdowns')):
+                message = f"For {insight.get('name')} that action_breakdowns are not espected in action_breakdowns"
+                raise Exception("Config validation error: " + message) from None
+
+        return True
+
+    def _check_values(self, default_value: List[str], custom_value: List[str]) -> bool:
+        for e in custom_value:
+            if e not in default_value:
+                logger.error(f"{e} does not appers in {default_value}")
+                return False
+        return True
+
+
+def expand_local_ref(schema, resolver=None, **kwargs):
+    resolver = resolver or RefResolver("", schema)
+    if isinstance(schema, MutableMapping):
+        if "$ref" in schema:
+            ref_url = schema.pop("$ref")
+            url, resolved_schema = resolver.resolve(ref_url)
+            schema.update(resolved_schema)
+        for key, value in schema.items():
+            schema[key] = expand_local_ref(value, resolver=resolver)
+        return schema
+    elif isinstance(schema, List):
+        return [expand_local_ref(item, resolver=resolver) for item in schema]
+
+    return schema

docs/integrations/sources/facebook-marketing.md

@@ -76,6 +76,12 @@ Facebook heavily throttles API tokens generated from Facebook Apps by default, m
 
 See Facebook's [documentation on rate limiting](https://developers.facebook.com/docs/marketing-api/overview/authorization/#access-levels) for more information on requesting a quota upgrade.
 
+## Custom Insights
+In order to retrieve specific fields from Facebook Ads Insights combined with other breakdowns, there is a mechanism to allow you to choose which fields and breakdowns to sync.
+It is highly recommended to follow the [documenation](https://developers.facebook.com/docs/marketing-api/insights/breakdowns), as there are limitations related to breakdowns. Some fields can not be requested and many others just work combined with specific fields, for example, the breakdown **app_id** is only supported with the **total_postbacks** field.
+By now, the only check done when setting up a source is to check if the fields, breakdowns and action breakdowns are within the ones provided by Facebook. This is, if you enter a good input, it's gonna be validated, but after, if the calls to Facebook API with those pareameters fails you will receive an error from the API.
+As a summary, custom insights allows to replicate only some fields, resulting in sync speed increase.
+
 #### Data type mapping
 
 | Integration Type | Airbyte Type | Notes |