example_ga_dag.py

# -*- coding: utf-8 -*-

from datetime import datetime, timedelta

from airflow import DAG
from airflow.operators import GoogleApiToS3Transfer

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2019, 6, 1),
    'retries': 2,
    'retry_delay': timedelta(minutes=1),
}

dag = DAG(
    dag_id='google_analytics_v1',
    default_args=default_args,
    catchup=True,
    schedule_interval="55 8 * * *"
)

with dag:
    google_analytics_reporting_to_s3 = GoogleApiToS3Transfer(
        task_id='google_analytics_reporting_to_s3',
        google_api_service_name='analytics',
        google_api_service_version='v4',
        google_api_endpoint_path='analyticsreporting.reports.batchGet',
        google_api_endpoint_params={
            'body': {
                'reportRequests': [{
                    'viewId':
                    "95198094",
                    'dateRanges': [{
                        'startDate': '{{ ds }}',
                        'endDate': '{{ ds }}'
                    }],
                    'metrics': [{
                        'expression': 'ga:users'
                    }]
                }]
            }
        },
        s3_destination_key='s3://google-analytics-louisguitton/' + '{{ ds }}' + '.json',
        dag=dag
    )