Creating asynchronous reports with report tasks in the Google Analytics Data API

Report tasks in the Google Analytics Data API (v1alpha) let you start a long-running asynchronous request to create a customized report of your Google Analytics event data. Instead of waiting for a report in a single synchronous call, you create a report task, check its readiness state, and then query the generated report once it's ready.

A report task created by one user can be listed and queried by all users with read access to the Google Analytics property. A generated report is available for 72 hours after it's ready. After this period, the report task resource and its contents are automatically deleted.

This notebook demonstrates how to use the google-analytics-data Python client library (AlphaAnalyticsDataClient) to work with report tasks:

  1. Configure the Colab environment
  2. Install and initialize the Python client
  3. Configure your GA property ID
  4. Create a report task
  5. Wait for the report task to become ready
  6. Retrieve the generated report
  7. List report tasks for a property
  8. Interactive report task builder

Step 0: Configure the Colab environment

To run this notebook in Google Colab, configure Application Default Credentials (ADC) with the https://www.googleapis.com/auth/analytics.readonly scope and a Google Cloud quota project that has the Google Analytics Data API enabled.

Choose an AUTH_METHOD in the form below based on your environment:

  • User Account (gcloud ADC) (default): Authenticates with your Google user account via gcloud auth application-default login and sets PROJECT_ID as the quota project. Sign in with an account that has access to your Google Analytics property.
  • Service Account Key JSON: Prompts you to upload a service account key .json file via google.colab.auth.authenticate_service_account(). Make sure the service account email has been granted Viewer access to the Google Analytics property.
  • Service Account Impersonation: Keyless authentication using --impersonate-service-account with SERVICE_ACCOUNT_EMAIL. Avoids requesting analytics.readonly on gcloud's default OAuth client ID (requires roles/iam.serviceAccountTokenCreator on the service account and Viewer access on the GA property).
  • Custom OAuth Client JSON: Prompts you to upload your own Desktop OAuth client_secret.json (create one in Cloud Console) and passes --client-id-file to gcloud (recommended if your organization blocks analytics.readonly on gcloud's default client ID; see ADC scope troubleshooting).

Need a Google Cloud project? If you don't already have a Google Cloud project with the Data API enabled, open the 1-click Enable the Google Analytics Data API console wizard. It will create a project or let you select an existing one and enable the API automatically. Then copy the project ID and enter it below.

Colab Authentication and Quota Project Setup

Step 1: Install and initialize the Python client

Install the google-analytics-data Python client library and initialize AlphaAnalyticsDataClient from google.analytics.data_v1alpha. The client automatically detects and uses the Application Default Credentials configured in your environment.

%pip install --upgrade -q google-analytics-data

from google.analytics import data_v1alpha

# Create a client using Application Default Credentials
client = data_v1alpha.AlphaAnalyticsDataClient()
print("AlphaAnalyticsDataClient initialized successfully.")

Step 2: Configure your GA4 property ID

Enter your numeric Google Analytics 4 Property ID below. You can find your Property ID in the Google Analytics UI under Admin > Property settings > Property details > Property ID.

# @title GA Property Settings
PROPERTY_ID = "YOUR_PROPERTY_ID"  # @param {type:"string"}

if PROPERTY_ID == "YOUR_PROPERTY_ID" or not PROPERTY_ID.strip():
  raise ValueError("Please provide a valid numeric GA Property ID.")

property_resource = f"properties/{PROPERTY_ID.strip()}"
print(f"Configured property resource: {property_resource}")

Step 3: Create a report task

To create a report task, call the create_report_task method with a ReportTask object. The task's report_definition describes the customized report, and its structure is similar to the report definition used by the Core Reporting methods: dimensions, metrics, date_ranges, filters, order_bys, offset, limit, and so on.

create_report_task returns a long-running operation. Calling operation.result() returns the newly created ReportTask, whose name field (such as properties/1234567/reportTasks/123) identifies the task in subsequent requests. At this point, the report is usually still in the CREATING state.

Scenario: Active users and sessions by country and device

Suppose you want to build a report of active users and sessions broken down by country and device category over the last 28 days, sorted by active users.

# Define the customized report
report_definition = data_v1alpha.ReportTask.ReportDefinition(
    dimensions=[
        data_v1alpha.Dimension(name="country"),
        data_v1alpha.Dimension(name="deviceCategory"),
    ],
    metrics=[
        data_v1alpha.Metric(name="activeUsers"),
        data_v1alpha.Metric(name="sessions"),
    ],
    date_ranges=[
        data_v1alpha.DateRange(start_date="28daysAgo", end_date="yesterday")
    ],
    order_bys=[
        data_v1alpha.OrderBy(
            metric=data_v1alpha.OrderBy.MetricOrderBy(metric_name="activeUsers"),
            desc=True,
        )
    ],
)

# Initialize request arguments
request = data_v1alpha.CreateReportTaskRequest(
    parent=property_resource,
    report_task=data_v1alpha.ReportTask(report_definition=report_definition),
)

# Make the request. The method returns a long-running operation.
operation = client.create_report_task(request=request)
report_task = operation.result()

# Save the report task name for subsequent requests
report_task_name = report_task.name
print(f"Report task name: {report_task_name}")
print(f"Initial state: {report_task.report_metadata.state.name}")

Step 4: Wait for the report task to become ready

It may take several minutes to generate a report after the create_report_task call. To check the readiness state of a report task, call the get_report_task method with the report task name.

The readiness status is returned in the report_metadata.state field. Once report generation is complete, the state changes from CREATING to ACTIVE (or to FAILED if the report can't be generated, in which case report_metadata.error_message explains why).

The report_metadata field also contains high-level information about the generated report:

  • begin_creating_time: The time when report creation started.
  • creation_quota_tokens_charged: The total quota tokens charged during report creation.
  • task_row_count: The number of rows in the generated report.
  • total_row_count: The total number of rows available in Google Analytics storage. If it's larger than task_row_count, you can create another report task with a different offset to retrieve the remaining rows.

Let's define a helper function that polls get_report_task until the report task leaves the CREATING state.

import time

State = data_v1alpha.ReportTask.ReportMetadata.State


def wait_for_report_task(
    name: str, poll_interval_seconds: int = 10, timeout_seconds: int = 900
) -> data_v1alpha.ReportTask:
  """Polls a report task until it is ACTIVE, and returns the ReportTask."""
  deadline = time.monotonic() + timeout_seconds
  while True:
    task = client.get_report_task(
        request=data_v1alpha.GetReportTaskRequest(name=name)
    )
    state = task.report_metadata.state
    print(f"{time.strftime('%H:%M:%S')} {name}: {state.name}")

    if state == State.ACTIVE:
      return task
    if state == State.FAILED:
      raise RuntimeError(
          f"Report task failed: {task.report_metadata.error_message}"
      )
    if time.monotonic() > deadline:
      raise TimeoutError(
          f"Report task is still {state.name} after {timeout_seconds} seconds."
          " Re-run this cell later to check again."
      )
    time.sleep(poll_interval_seconds)


def print_report_metadata(task: data_v1alpha.ReportTask) -> None:
  """Prints high-level information about a generated report."""
  metadata = task.report_metadata
  print(f"State: {metadata.state.name}")
  print(f"Began creating: {metadata.begin_creating_time}")
  print(f"Quota tokens charged: {metadata.creation_quota_tokens_charged}")
  print(f"Task row count: {metadata.task_row_count:,}")
  print(f"Total row count: {metadata.total_row_count:,}")


# Wait for the report task from Step 3 to become ready
report_task = wait_for_report_task(report_task_name)
print_report_metadata(report_task)

Step 5: Retrieve the generated report

Once the report task is ACTIVE, call the query_report_task method with the report task name to retrieve the generated report. The method returns an error if the report task isn't ACTIVE yet.

The QueryReportTaskResponse has the same shape as a Core Reporting response: dimension_headers, metric_headers, rows (each with dimension_values and metric_values), row_count, and metadata.

Use the optional offset and limit parameters to page through large reports. If unspecified, up to 10,000 rows are returned per request. The helper below pages through the entire report and converts it into a pandas.DataFrame.

from IPython.display import display
import pandas as pd


def query_report_task_to_dataframe(
    name: str, page_size: int = 10000
) -> pd.DataFrame:
  """Retrieves all rows of a generated report as a pandas DataFrame."""
  headers = []
  rows = []
  offset = 0
  while True:
    response = client.query_report_task(
        request=data_v1alpha.QueryReportTaskRequest(
            name=name, offset=offset, limit=page_size
        )
    )
    if not headers:
      headers = [h.name for h in response.dimension_headers] + [
          h.name for h in response.metric_headers
      ]
    rows.extend(
        [v.value for v in row.dimension_values]
        + [v.value for v in row.metric_values]
        for row in response.rows
    )
    offset += len(response.rows)
    if not response.rows or offset >= response.row_count:
      break

  print(
      f"Retrieved {len(rows):,} row(s). Currency: "
      f"{response.metadata.currency_code}, time zone: "
      f"{response.metadata.time_zone}"
  )
  return pd.DataFrame(rows, columns=headers)


# Retrieve the report generated in Step 4
df = query_report_task_to_dataframe(report_task_name)
display(df.head(20))

Step 6: List report tasks for a property

To list all report tasks for a property, call the list_report_tasks method. The client library returns a pager that automatically fetches subsequent pages as you iterate. Because report tasks are shared, the list includes tasks created by all users with access to the property during the last 72 hours.

# Make the request. The pager handles page tokens automatically.
pager = client.list_report_tasks(
    request=data_v1alpha.ListReportTasksRequest(parent=property_resource)
)

tasks_df = pd.DataFrame(
    [
        {
            "name": task.name,
            "state": task.report_metadata.state.name,
            "begin_creating_time": task.report_metadata.begin_creating_time,
            "dimensions": ", ".join(
                d.name for d in task.report_definition.dimensions
            ),
            "metrics": ", ".join(m.name for m in task.report_definition.metrics),
            "task_row_count": task.report_metadata.task_row_count,
        }
        for task in pager
    ]
)
display(tasks_df)

Step 7: Interactive report task builder

Use the interactive form cell below to create a custom report task for your GA4 property. Enter comma-separated dimension and metric API names, a date range, and a row limit. The cell creates the report task, waits for it to become ready, and displays the generated report.

# @title Create a Custom Report Task
dimensions = "date, sessionDefaultChannelGroup"  # @param {type:"string"}
metrics = "activeUsers, newUsers, engagementRate"  # @param {type:"string"}
start_date = "7daysAgo"  # @param {type:"string"}
end_date = "yesterday"  # @param {type:"string"}
row_limit = 10000  # @param {type:"integer"}


def _split(names: str) -> list[str]:
  return [name.strip() for name in names.split(",") if name.strip()]


custom_request = data_v1alpha.CreateReportTaskRequest(
    parent=property_resource,
    report_task=data_v1alpha.ReportTask(
        report_definition=data_v1alpha.ReportTask.ReportDefinition(
            dimensions=[data_v1alpha.Dimension(name=n) for n in _split(dimensions)],
            metrics=[data_v1alpha.Metric(name=n) for n in _split(metrics)],
            date_ranges=[
                data_v1alpha.DateRange(start_date=start_date, end_date=end_date)
            ],
            limit=row_limit,
        )
    ),
)

custom_task = client.create_report_task(request=custom_request).result()
custom_task = wait_for_report_task(custom_task.name)
print_report_metadata(custom_task)
display(query_report_task_to_dataframe(custom_task.name))

Additional resources