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

%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.")

# @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}")

# 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}")

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)

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))

# 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)

# @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))