0
# Apache Airflow Facebook Provider (Backport)
1

2
A backport provider package that enables Facebook Ads API integration for Apache Airflow 1.10.x installations. This package provides hooks for connecting to Facebook's Marketing API to extract advertising data, manage campaigns, and integrate Facebook Ads reporting into Airflow workflows with asynchronous job execution and comprehensive error handling.
3

4
## Package Information
5

6
- **Package Name**: apache-airflow-backport-providers-facebook
7
- **Package Type**: pypi
8
- **Language**: Python
9
- **Installation**: `pip install apache-airflow-backport-providers-facebook`
10
- **Python Support**: 3.6+
11
- **Dependencies**: `apache-airflow~=1.10`, `facebook-business>=6.0.2`
12

13
## Core Imports
14

15
```python
16
from airflow.providers.facebook.ads.hooks.ads import FacebookAdsReportingHook, JobStatus
17
```
18

19
## Basic Usage
20

21
```python
22
from airflow.providers.facebook.ads.hooks.ads import FacebookAdsReportingHook
23
from airflow.operators.python_operator import PythonOperator
24
from datetime import datetime, timedelta
25

26
def extract_facebook_data(**context):
27
    # Initialize the hook with connection ID and API version
28
    hook = FacebookAdsReportingHook(
29
        facebook_conn_id="facebook_default",
30
        api_version="v6.0"
31
    )
32
    
33
    # Define report parameters
34
    params = {
35
        "level": "ad",
36
        "date_preset": "yesterday",
37
        "time_increment": 1
38
    }
39
    
40
    # Define fields to extract
41
    fields = [
42
        "campaign_name",
43
        "campaign_id", 
44
        "ad_id",
45
        "clicks",
46
        "impressions",
47
        "spend",
48
        "cpc",
49
        "cpm"
50
    ]
51
    
52
    # Extract data using bulk reporting
53
    data = hook.bulk_facebook_report(
54
        params=params,
55
        fields=fields,
56
        sleep_time=5
57
    )
58
    
59
    # Process the returned AdsInsights objects
60
    for insight in data:
61
        print(f"Campaign: {insight['campaign_name']}, Clicks: {insight['clicks']}")
62
    
63
    return data
64

65
# Use in an Airflow DAG
66
facebook_extract_task = PythonOperator(
67
    task_id='extract_facebook_ads_data',
68
    python_callable=extract_facebook_data,
69
    dag=dag
70
)
71
```
72

73
## Connection Configuration
74

75
The hook requires an Airflow connection with the following configuration:
76

77
- **Connection ID**: `facebook_default` (or custom)
78
- **Connection Type**: Any (e.g., HTTP)
79
- **Extra**: JSON containing required Facebook API credentials:
80

81
```json
82
{
83
    "app_id": "your_facebook_app_id",
84
    "app_secret": "your_facebook_app_secret", 
85
    "access_token": "your_facebook_access_token",
86
    "account_id": "act_your_facebook_account_id"
87
}
88
```
89

90
## Capabilities
91

92
### Facebook Ads Reporting Hook
93

94
The main integration component for connecting to Facebook's Marketing API and extracting advertising data through asynchronous report generation.
95

96
```python { .api }
97
class FacebookAdsReportingHook(BaseHook):
98
    """
99
    Hook for the Facebook Ads API.
100
    
101
    Inherits from BaseHook to integrate with Airflow's connection management system.
102
    Validates connection configuration on initialization and provides methods for
103
    asynchronous Facebook Ads data extraction.
104
    
105
    Args:
106
        facebook_conn_id (str): Airflow Facebook Ads connection ID. Defaults to "facebook_default"
107
        api_version (str): The version of Facebook API. Defaults to "v6.0"
108
        
109
    Attributes:
110
        client_required_fields (List[str]): Required fields for Facebook API connection
111
                                          ["app_id", "app_secret", "access_token", "account_id"]
112
    """
113
    
114
    def __init__(
115
        self,
116
        facebook_conn_id: str = "facebook_default",
117
        api_version: str = "v6.0",
118
    ) -> None: ...
119
    
120
    def bulk_facebook_report(
121
        self,
122
        params: Dict[str, Any],
123
        fields: List[str],
124
        sleep_time: int = 5,
125
    ) -> List[AdsInsights]:
126
        """
127
        Pulls data from the Facebook Ads API using asynchronous reporting.
128

129
        Args:
130
            params (Dict[str, Any]): Parameters that determine the query for Facebook Ads API.
131
                Must follow Facebook Marketing API Insights Parameters format.
132
                Common parameters include:
133
                - level (str): Report level - "account", "campaign", "adset", "ad"
134
                - date_preset (str): Date range preset - "today", "yesterday", "last_7_days", 
135
                  "last_14_days", "last_28_days", "this_month", "last_month", etc.
136
                - time_range (Dict): Custom date range with "since" and "until" keys in YYYY-MM-DD format
137
                  Example: {"since": "2023-01-01", "until": "2023-01-31"}
138
                - time_increment (int): Time increment for date breakdown (1=daily, 7=weekly, "monthly")
139
                - breakdowns (List[str]): Breakdown dimensions like ["age", "gender", "placement", "device_platform"]
140
                - action_breakdowns (List[str]): Action breakdown dimensions like ["action_type", "action_target_id"]
141
            fields (List[str]): List of fields to obtain from Facebook Ads Insights API.
142
                Must be valid Facebook Ads Insights fields. Common fields include:
143
                - Identifiers: "campaign_name", "campaign_id", "adset_name", "adset_id", "ad_name", "ad_id"
144
                - Metrics: "impressions", "clicks", "spend", "reach", "frequency"
145
                - Calculated: "cpc", "cpm", "ctr", "cpp", "cost_per_unique_click"
146
                - Conversions: "conversions", "cost_per_conversion", "conversion_rate"
147
                - Video: "video_plays", "video_p25_watched_actions", "video_p50_watched_actions"
148
            sleep_time (int): Time to sleep between async job status checks. Defaults to 5 seconds.
149
                Increase for large reports to reduce API calls.
150

151
        Returns:
152
            List[AdsInsights]: Facebook Ads API response as list of AdsInsights objects.
153
                Each AdsInsights object is a dictionary-like object containing the requested
154
                fields as key-value pairs (e.g., {"campaign_name": "My Campaign", "clicks": "150"}).
155
            
156
        Raises:
157
            AirflowException: If any required connection fields (app_id, app_secret, access_token, 
158
                            account_id) are missing from the connection's extra configuration, or if
159
                            the async report job fails or is skipped by Facebook's API.
160
        """
161

162
    @cached_property
163
    def facebook_ads_config(self) -> Dict:
164
        """
165
        Gets Facebook ads connection configuration from Airflow connections.
166
        
167
        Retrieves connection details from Airflow's meta database using the configured
168
        facebook_conn_id and validates that all required fields are present in the
169
        connection's extra_dejson configuration.
170
        
171
        Required connection fields:
172
            - app_id (str): Facebook App ID
173
            - app_secret (str): Facebook App Secret  
174
            - access_token (str): Facebook Access Token
175
            - account_id (str): Facebook Account ID (format: "act_12345")
176
        
177
        Returns:
178
            Dict: Configuration dictionary containing app_id, app_secret, access_token, and account_id.
179
            
180
        Raises:
181
            AirflowException: If any required fields (app_id, app_secret, access_token, account_id) 
182
                            are missing from the connection's extra configuration.
183
        """
184
```
185

186
### Job Status Enumeration
187

188
Enumeration of available status options for Facebook async task monitoring.
189

190
```python { .api }
191
class JobStatus(Enum):
192
    """Available options for facebook async task status"""
193
    
194
    COMPLETED = 'Job Completed'
195
    STARTED = 'Job Started'  
196
    RUNNING = 'Job Running'
197
    FAILED = 'Job Failed'
198
    SKIPPED = 'Job Skipped'
199
```
200

201
## Types
202

203
```python { .api }
204
from facebook_business.adobjects.adsinsights import AdsInsights
205

206
class AdsInsights:
207
    """
208
    Facebook Ads Insights object from facebook-business SDK.
209
    
210
    Dictionary-like object containing Facebook Ads data with requested fields
211
    as key-value pairs. Can be accessed like a dictionary:
212
    
213
    Example:
214
        insight['campaign_name']  # Returns campaign name string
215
        insight['clicks']         # Returns clicks count as string
216
        insight.get('spend', '0') # Returns spend with default fallback
217
    """
218
```
219

220
## Error Handling
221

222
The hook implements comprehensive error handling for Facebook API operations:
223

224
- **Missing Configuration**: Raises `AirflowException` with message "{missing_keys} fields are missing" 
225
  if any required connection fields are missing (`app_id`, `app_secret`, `access_token`, `account_id`)
226
- **Failed Async Jobs**: Raises `AirflowException` with message "{async_status}. Please retry." 
227
  if async report jobs fail or are skipped by Facebook's API
228
- **API Errors**: Propagates Facebook Business SDK exceptions for API-level errors such as:
229
  - Invalid authentication credentials
230
  - Insufficient permissions for requested data
231
  - Invalid parameter values or field names
232
  - Rate limiting and quota exceeded errors
233

234
## Common Use Cases
235

236
### Daily Campaign Performance Reports
237

238
```python
239
def get_daily_campaign_performance():
240
    hook = FacebookAdsReportingHook()
241
    
242
    params = {
243
        "level": "campaign",
244
        "date_preset": "yesterday"
245
    }
246
    
247
    fields = [
248
        "campaign_name",
249
        "impressions", 
250
        "clicks",
251
        "spend",
252
        "cpc",
253
        "cpm"
254
    ]
255
    
256
    return hook.bulk_facebook_report(params, fields)
257
```
258

259
### Custom Date Range with Breakdowns
260

261
```python
262
def get_ad_performance_with_breakdowns():
263
    hook = FacebookAdsReportingHook()
264
    
265
    params = {
266
        "level": "ad",
267
        "time_range": {
268
            "since": "2023-01-01",
269
            "until": "2023-01-31"
270
        },
271
        "breakdowns": ["age", "gender"],
272
        "time_increment": 1
273
    }
274
    
275
    fields = [
276
        "ad_name",
277
        "impressions",
278
        "clicks", 
279
        "conversions",
280
        "spend"
281
    ]
282
    
283
    return hook.bulk_facebook_report(params, fields, sleep_time=10)
284
```
285

286
## API References
287

288
- **Facebook Marketing APIs**: https://developers.facebook.com/docs/marketing-apis/
289
- **Insights Parameters**: https://developers.facebook.com/docs/marketing-api/insights/parameters/v6.0
290
- **Available Fields**: https://developers.facebook.com/docs/marketing-api/insights/fields/v6.0