# Google Analytics connector Source: /docs/connectors/google-analytics-data-api.md Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See [Custom API](/docs/connectors/custom-api) for the shared SDK pattern and document-upload workflow. ## What this connector is for [#what-this-connector-is-for] Web and app analytics for events, traffic, audiences, and conversion reporting. A connector owns access to the source. Separate syncs shape selected resources into tables for analysis and downstream transformations using the [Python SDK](/docs/reference/python/connectors-and-syncs). ## Data and tables you can pull [#data-and-tables-you-can-pull] These are representative dataset families to consider, not a claim that the worked example implements every resource. Access, fields, and availability depend on the source account, product edition, and permissions. Use the [vendor reference](https://developers.google.com/analytics/devguides/reporting/data/v1) to select concrete resources. | Suggested table | One row represents | | --- | --- | | `daily_traffic` | a daily traffic | | `event_reports` | a event report | | `conversion_reports` | a conversion report | | `audience_reports` | a audience report | | `properties` | a property | ### Relationships and extraction considerations [#relationships-and-extraction-considerations] Report rows represent the requested combination of dimensions, metrics, date range, and property. They are aggregates rather than a raw event dump. Different metric/dimension combinations are not always compatible. ## Connecting to Google Analytics [#connecting-to-google-analytics] Uses a Google OAuth token with analytics.readonly and access to the numeric GA4 property. This is an aggregated report, not a raw-event export. Additional resources may require additional scopes. Connection settings belong to the connector; dataset filters and resource selection belong to each sync. See the [CLI workflow](/docs/reference/cli/overview) for registration and configuration. ### Connection settings used by the example [#connection-settings-used-by-the-example] | Setting | Requirement | Purpose | | --- | --- | --- | | `access_token` | Required; secret | Platform-supplied access token. Refresh or rotate before expiry. | | `property_id` | Required | Property Id | ### requirements.in [#requirementsin] Save these dependencies in `requirements.in` alongside `connector.py`; the example includes packages used by its sync. Install the published `vortexify` package for local development. Keep sync-specific dependencies in the sync's own `requirements.in` when splitting the implementation into separate resources. ```text vortexify==0.1.1 httpx>=0.27,<1 pandas>=2,<3 ``` No separately installed database driver is needed by this HTTP example. ## Worked example: sync `daily_users` [#worked-example-sync-daily_users] **daily_users** — Date/country activeUsers and sessions report for a fixed date window. GA4 thresholds, quotas, and reporting data limits apply. The complete files below implement only this example resource. They illustrate the current SDK signatures, a separate preview, bounded DataFrame output, and source-specific extraction. They do not establish complete source coverage. Live credentials and platform ingestion have not been used to validate this guide. With default paths, place the connector files in `connectors/google-analytics-data-api/` and sync files in `connectors/google-analytics-data-api/syncs/daily_users/`. Use your project configuration if it specifies different paths. ### Connection schema: `config.json` [#connection-schema-configjson] This file contains a JSON Schema, not actual credential values. Provide those through platform configuration. ```json { "type": "object", "properties": { "access_token": { "type": "string", "title": "Access Token", "description": "Platform-supplied access token. Refresh or rotate before expiry.", "minLength": 1, "x-vortexify-secret": true, "order": 0 }, "property_id": { "type": "string", "minLength": 1, "title": "Property Id", "order": 1 } }, "required": [ "access_token", "property_id" ] } ``` ### Connector: `connector.py` [#connector-connectorpy] ```python import httpx from vortexify import Column, Connector, DiscoveredTable class GoogleAnalyticsDataApiConnector(Connector): vendor_domain = "google.com" def client(self): if self._client is None or self._client.is_closed: self._client = httpx.Client( base_url="https://analyticsdata.googleapis.com/v1beta/", headers={ "Authorization": "Bearer " + self.config["access_token"], "Accept": "application/json", }, timeout=60, follow_redirects=False, ) return self._client def check(self) -> bool: response = self.client().get( "properties/" + self.config["property_id"] + "/metadata" ) response.raise_for_status() return True def discover(self) -> list[DiscoveredTable]: return [ DiscoveredTable( name="daily_users", schema=[ Column(name="date", type="string"), Column(name="country", type="string"), Column(name="active_users", type="integer"), Column(name="sessions", type="integer"), Column(name="document", type="string"), ], ) ] def close(self) -> None: if self._client is not None: try: self._client.close() finally: self._client = None ``` ### Sync schema: `syncs/daily_users/config.json` [#sync-schema-syncsdaily_usersconfigjson] ```json { "type": "object", "properties": { "start_date": { "type": "string", "title": "Start Date", "description": "YYYY-MM-DD inclusive.", "minLength": 1, "order": 0 }, "end_date": { "type": "string", "title": "End Date", "description": "YYYY-MM-DD inclusive.", "minLength": 1, "order": 1 } }, "required": [ "start_date", "end_date" ] } ``` ### Sync: `syncs/daily_users/sync.py` [#sync-syncsdaily_userssyncpy] ```python from collections.abc import Iterator from math import isfinite import json import pandas as pd from vortexify import Sync class GoogleAnalyticsDataApiDailyUsersSync(Sync): def preview(self) -> pd.DataFrame: cfg = self.connector.config page_size = 100 path = "properties/" + cfg["property_id"] + ":runReport" body = { "dateRanges": [ { "startDate": self.config["start_date"], "endDate": self.config["end_date"], } ], "dimensions": [{"name": "date"}, {"name": "country"}], "metrics": [{"name": "activeUsers"}, {"name": "sessions"}], "limit": page_size, "offset": 0, "orderBys": [ {"dimension": {"dimensionName": "date"}}, {"dimension": {"dimensionName": "country"}}, ], } response = self.client.post(path, json=body) response.raise_for_status() payload = response.json() records = payload.get("rows", []) if not isinstance(records, list) or any( (not isinstance(record, dict) for record in records) ): raise ValueError( "Unexpected API record shape; refusing a partial successful sync" ) records = records[:100] rows = [] for record in records: row = { "date": record["dimensionValues"][0]["value"], "country": record["dimensionValues"][1]["value"], "active_users": int(record["metricValues"][0]["value"]), "sessions": int(record["metricValues"][1]["value"]), "document": json.dumps(record, ensure_ascii=False, default=str), } rows.append( { key: ( json.dumps(value, ensure_ascii=False, default=str) if isinstance(value, (dict, list)) else value ) for key, value in row.items() } ) return pd.DataFrame( rows, columns=["date", "country", "active_users", "sessions", "document"], dtype=object, ) def extract( self, *, avg_row_size_bytes: float | None = None ) -> Iterator[pd.DataFrame]: memory_mib = self.compute.memory if self.compute else None memory_mib = memory_mib if memory_mib is not None and memory_mib > 0 else 1024 target_bytes = min(16 * 1024 * 1024, max(1, memory_mib * 1024 * 1024 // 20)) batch_size = 100 if avg_row_size_bytes is not None: if not isfinite(avg_row_size_bytes) or avg_row_size_bytes <= 0: raise ValueError("avg_row_size_bytes must be finite and positive") batch_size = max(1, int(target_bytes / max(1.0, avg_row_size_bytes) / 4)) batch_size = min(batch_size, 10000) cfg = self.connector.config page_size = batch_size path = "properties/" + cfg["property_id"] + ":runReport" body = { "dateRanges": [ { "startDate": self.config["start_date"], "endDate": self.config["end_date"], } ], "dimensions": [{"name": "date"}, {"name": "country"}], "metrics": [{"name": "activeUsers"}, {"name": "sessions"}], "limit": page_size, "offset": 0, "orderBys": [ {"dimension": {"dimensionName": "date"}}, {"dimension": {"dimensionName": "country"}}, ], } seen_pages = set() while True: response = self.client.post(path, json=body) response.raise_for_status() payload = response.json() records = payload.get("rows", []) if not isinstance(records, list) or any( (not isinstance(record, dict) for record in records) ): raise ValueError( "Unexpected API record shape; refusing a partial successful sync" ) rows = [] for record in records: row = { "date": record["dimensionValues"][0]["value"], "country": record["dimensionValues"][1]["value"], "active_users": int(record["metricValues"][0]["value"]), "sessions": int(record["metricValues"][1]["value"]), "document": json.dumps(record, ensure_ascii=False, default=str), } rows.append( { key: ( json.dumps(value, ensure_ascii=False, default=str) if isinstance(value, (dict, list)) else value ) for key, value in row.items() } ) for start in range(0, len(rows), batch_size): yield pd.DataFrame( rows[start : start + batch_size], columns=["date", "country", "active_users", "sessions", "document"], dtype=object, ) body["offset"] += len(records) if body["offset"] >= payload.get("rowCount", 0): break if not records: raise ValueError("Empty GA4 page before rowCount") ``` ### Running and extending the example [#running-and-extending-the-example] Register the connector before the sync and commit each resource separately: ```sh vortexify create connector google-analytics-data-api vortexify create sync google-analytics-data-api/daily_users ``` Edit the files, inspect `vortexify status`, and use the [CLI guide](/docs/reference/cli/overview) to commit and configure them. Registration alone does not run extraction or deploy the connector. Preview must run with a fresh connector and disposable state. The extraction implementation determines full/incremental behavior; merely setting a mode does not implement change capture. API/driver failures propagate unless explicitly handled in the code. Check source limits and single-record/response sizes as well as output batch size. ## Sources and further reading [#sources-and-further-reading] - [Vendor API and data reference](https://developers.google.com/analytics/devguides/reporting/data/v1) — Resource definitions, authentication, fields, and source-specific limits. Check the selected resource's reference when extending the example. - [Vortexify connector and sync contracts](/docs/reference/python/connectors-and-syncs) — Client ownership, discovery, preview isolation, and DataFrame extraction. - [Vortexify CLI workflow](/docs/reference/cli/overview) — Project paths, resource registration, source commits, and credential configuration.