Implementation:Mage ai Mage ai Google Analytics Source
| Knowledge Sources | |
|---|---|
| Domains | Data_Integration, Google_Analytics, Source_Connector |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Concrete tool for extracting data from Google Analytics 4 properties using the Data API v1beta provided by the Mage integrations source connector framework.
Description
The GoogleAnalytics source connector extends the base Source class to implement data extraction from Google Analytics 4 properties via the GA Data API v1beta. It initializes a GoogleAnalyticsConnection in the constructor using a property_id and a path to a service account credentials JSON file. Discovery calls the GA metadata endpoint to retrieve all available dimensions and metrics for the property. Dimensions are mapped to string type, while metrics are mapped based on their MetricType: TYPE_INTEGER becomes integer, TYPE_FLOAT becomes number, and all others become string, each with a nullable prefix. The discovered stream is named google_analytics_custom. The load_data() method extracts selected columns, separates them into dimensions and metrics (using predefined constant lists), and fetches data with pagination using a row limit of 10,000 per request. Date ranges are controlled by start_date and optional end_date configuration. The get_forced_replication_method() always returns full-table replication. The test_connection() method calls the connection's connect() method to verify credentials and access.
Usage
Use this source connector when building a Mage data pipeline that needs to extract analytics data from a Google Analytics 4 property. Configure with property_id, path_to_credentials_json_file, start_date, and optionally end_date.
Code Reference
Source Location
- Repository: mage-ai
- File: mage_integrations/mage_integrations/sources/google_analytics/__init__.py
- Lines: 1-128
Signature
class GoogleAnalytics(Source):
ROW_LIMIT = 10000
def __init__(self, **kwargs):
...
def discover(self, streams: List[str] = None) -> Catalog:
...
def load_data(self, stream, bookmarks: Dict = None, query: Dict = {}, **kwargs) -> Generator[List[Dict], None, None]:
...
def get_forced_replication_method(self, stream_id):
...
def test_connection(self):
...
Import
from mage_integrations.sources.google_analytics import GoogleAnalytics
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| config | dict | Yes | Configuration dictionary with Google Analytics property and credentials |
| catalog | Catalog | No | Singer catalog specifying streams to extract |
| state | dict | No | Previous sync state for incremental extraction |
Configuration Parameters
| Name | Type | Required | Description |
|---|---|---|---|
| property_id | str | Yes | Google Analytics 4 property ID |
| path_to_credentials_json_file | str | Yes | File path to Google service account credentials JSON file |
| start_date | str | Yes | Start date for the analytics report (e.g., 2024-01-01) |
| end_date | str | No | End date for the analytics report; defaults to today if omitted |
Outputs
| Name | Type | Description |
|---|---|---|
| catalog | Catalog | Single stream 'google_analytics_custom' with all available dimensions and metrics (from discover()) |
| records | Generator[List[Dict]] | Batches of analytics rows with selected dimensions and metrics (from load_data()) |
Usage Examples
from mage_integrations.sources.google_analytics import GoogleAnalytics
config = {
"property_id": "123456789",
"path_to_credentials_json_file": "/path/to/service_account.json",
"start_date": "2024-01-01",
"end_date": "2024-12-31",
}
source = GoogleAnalytics(config=config)
# Discover available streams
catalog = source.discover()
# Test connection
source.test_connection()