Docs
Connectors

Google Search Console connector

Search performance and website indexing data for SEO reporting.

View Markdown

Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See Custom API for the shared SDK pattern and document-upload workflow.

What this connector is for

Search performance and website indexing data for SEO reporting.

A connector owns access to the source. Separate syncs shape selected resources into tables for analysis and downstream transformations using the Python SDK.

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 to select concrete resources.

Suggested tableOne row represents
sitesa site
search_analyticsa requested dimension/date group
sitemapsa sitemap
url_inspection_resultsa url inspection result

Relationships and extraction considerations

Search-analytics rows are grouped by the requested dimensions and date window. API result limits mean a report should not be described as an exhaustive log of every search event.

Connecting to Google Search Console

Uses a Google OAuth token with webmasters.readonly and access to the verified property. Search Analytics data has vendor reporting limits.

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 for registration and configuration.

Connection settings used by the example

SettingRequirementPurpose
access_tokenRequired; secretPlatform-supplied access token. Refresh or rotate before expiry.
site_urlRequiredVerified property URL or sc-domain:example.com.

requirements.in

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.

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 search_analytics

search_analytics — Date/query search performance report in an explicit window. Google returns top rows and applies reporting limits; this is not an exhaustive event export.

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-search-console/ and sync files in connectors/google-search-console/syncs/search_analytics/. Use your project configuration if it specifies different paths.

Connection schema: config.json

This file contains a JSON Schema, not actual credential values. Provide those through platform configuration.

{
  "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
    },
    "site_url": {
      "type": "string",
      "title": "Site URL",
      "description": "Verified property URL or sc-domain:example.com.",
      "minLength": 1,
      "order": 1
    }
  },
  "required": [
    "access_token",
    "site_url"
  ]
}

Connector: connector.py

import httpx
from vortexify import Column, Connector, DiscoveredTable


class GoogleSearchConsoleConnector(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://www.googleapis.com/webmasters/v3/",
                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("sites")
        response.raise_for_status()
        return True

    def discover(self) -> list[DiscoveredTable]:
        return [
            DiscoveredTable(
                name="search_analytics",
                schema=[
                    Column(name="keys", type="json"),
                    Column(name="clicks", type="number"),
                    Column(name="impressions", type="number"),
                    Column(name="ctr", type="number"),
                    Column(name="position", type="number"),
                    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/search_analytics/config.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/search_analytics/sync.py

from collections.abc import Iterator
from math import isfinite
from urllib.parse import quote
import json
import pandas as pd
from vortexify import Sync


class GoogleSearchConsoleSearchAnalyticsSync(Sync):

    def preview(self) -> pd.DataFrame:
        cfg = self.connector.config
        page_size = 100
        path = "sites/" + quote(cfg["site_url"], safe="") + "/searchAnalytics/query"
        body = {
            "startDate": self.config["start_date"],
            "endDate": self.config["end_date"],
            "dimensions": ["date", "query"],
            "rowLimit": page_size,
            "startRow": 0,
        }
        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 = {
                "keys": record.get("keys"),
                "clicks": record.get("clicks"),
                "impressions": record.get("impressions"),
                "ctr": record.get("ctr"),
                "position": record.get("position"),
                "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=["keys", "clicks", "impressions", "ctr", "position", "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, 25000)
        cfg = self.connector.config
        page_size = batch_size
        path = "sites/" + quote(cfg["site_url"], safe="") + "/searchAnalytics/query"
        body = {
            "startDate": self.config["start_date"],
            "endDate": self.config["end_date"],
            "dimensions": ["date", "query"],
            "rowLimit": page_size,
            "startRow": 0,
        }
        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 = {
                    "keys": record.get("keys"),
                    "clicks": record.get("clicks"),
                    "impressions": record.get("impressions"),
                    "ctr": record.get("ctr"),
                    "position": record.get("position"),
                    "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=[
                        "keys",
                        "clicks",
                        "impressions",
                        "ctr",
                        "position",
                        "document",
                    ],
                    dtype=object,
                )
            if len(records) < page_size:
                break
            body["startRow"] += len(records)

Running and extending the example

Register the connector before the sync and commit each resource separately:

vortexify create connector google-search-console
vortexify create sync google-search-console/search_analytics

Edit the files, inspect vortexify status, and use the CLI guide 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

Google Drive connector

Cloud file repository with file metadata, folder navigation, permissions, and content.

Google Sheets connector

Spreadsheet platform for tabular data, worksheet updates, and reporting.

On this page

What this connector is forData and tables you can pullRelationships and extraction considerationsConnecting to Google Search ConsoleConnection settings used by the examplerequirements.inWorked example: sync search_analyticsConnection schema: config.jsonConnector: connector.pySync schema: syncs/search_analytics/config.jsonSync: syncs/search_analytics/sync.pyRunning and extending the exampleSources and further reading