Official Python client for Supermetrics
- PyPI package: https://pypi.org/project/supermetrics/
- Free software: Apache License v2
- Type-safe Python client generated from OpenAPI specification
- Dual sync/async support via separate Client classes
- Fully typed request and response models, generated from the spec as
attrsclasses - Comprehensive API coverage: login links (including update), logins (including account listing and revocation), accounts, queries, DWH transfers (including clone and batch create) and transfer runs, DWH destinations (including batch secret rotation), DWH table groups (list, export, import, edit), DWH backfills, custom fields, data blending, account tags, Connector Builder
- Custom exception hierarchy with HTTP status code mapping
- Resource-based API organization
- API key, OAuth bearer token, and dynamic token provider authentication
- Per-request authorization, header, and timeout overrides on a shared connection pool
with_raw_responseaccess to HTTP status codes, headers, and raw payloads- Automatic routing of Data Warehouse calls to their own host, from one pooled client
pip install supermetricsfrom supermetrics import SupermetricsClient
# Initialize client
client = SupermetricsClient(api_key="your_api_key")
# Create login link for data source authentication
link = client.login_links.create(ds_id="GAWA", description="My Analytics Authentication")
# Get login details after user authenticates
login = client.logins.get(login_id=link.login_id)
# List available accounts
accounts = client.accounts.list(ds_id="GAWA", login_usernames=login.username)
# Execute query
result = client.queries.execute(
ds_id="GAWA",
ds_accounts=[accounts[0].account_id],
fields=["Date", "Sessions", "Users"],
start_date="2024-01-01",
end_date="2024-01-07",
)
print(f"Retrieved {len(result.data)} rows")A login link starts the OAuth flow; once someone completes it a login exists, and that login
can reach a set of data source accounts. Beyond create, get, list and close, the SDK
lists a login's accounts, revokes a login, and updates a link's description.
from supermetrics import SupermetricsClient
client = SupermetricsClient(api_key="your_api_key")
# The accounts a login can reach. Paginated: offset and limit are always sent (0 and 100 by
# default), and the total rides in meta on the raw response, not in the returned list.
accounts = client.logins.get_accounts("login_abc123", offset=0, limit=100)
for account in accounts:
print(account.account_id, account.name, account.group)
response = client.with_raw_response.logins.get_accounts("login_abc123")
print(response.json_body["meta"]["paginate"]["total"])
# update() changes only the description — a link's data source, expiry and redirect are
# fixed at creation, so those are set through create(), not here.
link = client.login_links.update("link_123", "Q4 Analytics setup")
print(link.description) # "Q4 Analytics setup"
# revoke() is a DELETE that answers 200 with a bool, not an empty 204. True means the
# login's OAuth credentials were invalidated; bind a fresh login to restore access.
print(client.logins.revoke("login_abc123")) # TrueAn account tag groups data source accounts from across a team's connections under one
reusable label. A tag is addressed by name, the slug the server assigns it; the label
you choose is display_name.
from supermetrics import SupermetricsClient
client = SupermetricsClient(api_key="your_api_key")
# create() takes no name — the server assigns the slug and hands it back.
tag = client.account_tags.create(
team_id=12345,
display_name="EMEA paid media",
color="#112233",
data_sources=[{"data_source_id": "AW", "accounts": [{"account_id": "123-456-7890"}]}],
)
print(tag.name) # "a1b2c3d" — what every later call addresses this tag by
# list() summarises membership as counts; get() returns the membership itself.
for overview in client.account_tags.list(team_id=12345):
print(overview.display_name, overview.data_source_count, overview.account_count)
detail = client.account_tags.get(team_id=12345, name=tag.name)
print([selection["data_source_id"] for selection in detail.data_sources])
# update() renames and recolours, and does nothing else. Both fields are required, and
# it cannot move accounts — that is what add_accounts() and remove_accounts() are for.
client.account_tags.update(team_id=12345, name=tag.name, display_name="EMEA paid", color="#445566")
client.account_tags.add_accounts(
team_id=12345,
name=tag.name,
data_sources=[{"data_source_id": "FB", "accounts": [{"account_id": "act_99"}]}],
)
# delete() returns a bool, not None: deletion is idempotent upstream, so deleting a tag
# that is already gone is a success answering False rather than a 404. No operation in
# this domain answers 404 at all.
print(client.account_tags.delete(team_id=12345, name=tag.name)) # True
print(client.account_tags.delete(team_id=12345, name=tag.name)) # Falsefrom supermetrics import SupermetricsClient
client = SupermetricsClient(api_key="your_api_key")
# List connectors
connectors = client.connector_builder.list(team_id=12345)
# Create a connector
created = client.connector_builder.create(
team_id=12345, title="My Custom Connector", description="Fetches data from a custom API"
)
connector_id = created.connector_identifier
# Manage secrets
client.connector_builder_secrets.create(
team_id=12345, connector_identifier=connector_id, secret_name="api_key", secret_value="sk-secret-value"
)
# View execution logs
logs = client.connector_builder_logs.list(team_id=12345, connector_identifier=connector_id)Custom fields are calculated dimensions and metrics defined per team. Each carries a
definition: an ordered pipeline of function, lookup, and condition steps.
from supermetrics import DefinitionValue, FunctionArgument, FunctionStep, SupermetricsClient
client = SupermetricsClient(api_key="your_api_key")
# Discover which functions, rules, and data types the team is allowed to use
metadata = client.custom_fields.get_metadata(team_id=12345)
print([function.name for function in metadata.functions.items])
# Create a one-step field. field_type is "dim" or "met" and is fixed at creation.
field = client.custom_fields.create(
team_id=12345,
display_name="Platform (upper)",
field_type="dim",
data_type="string.text.value",
definition=[
FunctionStep(
type_="function",
name="upper_case",
arguments=[
FunctionArgument(name="value", value=DefinitionValue(type_="data_source_field", value="platform"))
],
)
],
)
# Read-modify-write. The definition comes back wrapped and is sent bare, so a round
# trip reads `.definition.items`. update() takes no field_type and replaces the whole
# object — anything you omit reverts to unset.
current = client.custom_fields.get(team_id=12345, custom_field_id=field.id)
client.custom_fields.update(
team_id=12345,
custom_field_id=field.id,
display_name="Platform (upper), revised",
data_type=current.data_type,
definition=current.definition.items,
description=current.description,
)
# list() returns the page only; the pagination rides in `meta` on the raw response,
# and total_count appears only when you ask for it.
response = client.with_raw_response.custom_fields.list(team_id=12345, include_total_count=True)
print(f"{len(response.data)} of {response.json_body['meta']['pagination']['total_count']} fields")
client.custom_fields.delete(team_id=12345, custom_field_id=field.id)A blend combines several data sources into one queryable table — "union" stacks their
rows, "join" joins them on shared fields. Sources being created have no id yet, so each
is named by a temporary eight-character blend_data_source_key that the field and join
references point at.
from supermetrics import (
BlendConfig,
BlendDatasourceFieldRef,
BlendedDataSourceInput,
BlendField,
SupermetricsClient,
)
client = SupermetricsClient(api_key="your_api_key")
# All five of these arguments are required, even though three are usually empty:
# upstream marks them required-but-nullable.
source = BlendedDataSourceInput(
data_source_id="GA4",
blend_data_source_id=None,
blend_data_source_key="abcd1234",
report_type=None,
report_type_settings=[],
display_name="Google Analytics 4",
)
# Collections go out as bare lists. They come back wrapped in `.items` — at every level.
config = BlendConfig(
fields=[
BlendField(
blend_field_name="impressions",
blend_field_display_name="Impressions",
blend_datasource_fields=[
BlendDatasourceFieldRef(
blend_data_source_key="abcd1234",
datasource_field_name="Impressions",
field_source="standard",
)
],
)
],
)
# create() answers 201 and assigns the ids. blend_type is fixed here and forever.
blend = client.blends.create(
team_id=12345,
display_name="GA4 impressions",
blend_type="union",
blended_data_sources=[source],
config=config,
)
print(blend.blend_id, blend.type_)
# list() returns summaries: no config, and a reduced source shape. This endpoint is not
# paginated, so one call returns every matching blend.
for summary in client.blends.list(team_id=12345, blend_type="union"):
print(summary.blend_id, summary.display_name)
# A read-modify-write has to rebuild the request objects: the response wraps its
# collections and drops blend_data_source_key, so it cannot be resent as-is. Existing
# sources are addressed by blend_data_source_id from here on. update() takes no
# blend_type and replaces the whole object: the required fields are resent in full, so a
# source or field you leave out is dropped.
current = client.blends.get(team_id=12345, blend_id=blend.blend_id)
existing = BlendedDataSourceInput(
data_source_id=current.blended_data_sources.items[0].data_source_id,
blend_data_source_id=current.blended_data_sources.items[0].blend_data_source_id,
blend_data_source_key=None,
report_type=None,
report_type_settings=[],
)
client.blends.update(
team_id=12345,
blend_id=blend.blend_id,
display_name="GA4 impressions, revised",
blended_data_sources=[existing],
config=BlendConfig(
fields=[
BlendField(
blend_field_name="impressions",
blend_field_display_name="Impressions",
blend_datasource_fields=[
BlendDatasourceFieldRef(
blend_data_source_id=existing.blend_data_source_id,
datasource_field_name="Impressions",
field_source="standard",
)
],
)
],
),
)
client.blends.delete(team_id=12345, blend_id=blend.blend_id)Read-only access to a team's identity and its members. Both calls are served by the core API host.
from supermetrics import SupermetricsClient
client = SupermetricsClient(api_key="your_api_key")
# Fetch a team's identity
team = client.teams.get(team_id=12345)
print(f"{team.team_id}: {team.name} ({team.status})")
# List the team's members
for user in client.teams.list_users(team_id=12345):
print(f"{user.email} ({user.role})")Transfers, transfer runs and backfills are served by the Data Warehouse API on a separate host. The SDK routes them there automatically, so an ordinary client reaches everything.
from datetime import UTC, datetime
from supermetrics import SupermetricsClient
client = SupermetricsClient(api_key="your_api_key")
# List transfers for a team
for transfer in client.transfers.list(team_id=12345):
print(f"{transfer.dwh_transfer_id}: {transfer.display_name} ({transfer.state})")
# Inspect one transfer's configuration
transfer = client.transfers.get(team_id=12345, transfer_id=36091)
print(f"Schedule: {transfer.schedule}")
# Dry-run a configuration before creating it. An invalid configuration comes back
# as a result, not an exception — that is the point of a validation endpoint.
from supermetrics._generated.supermetrics_api_client.models.transfer_account import TransferAccount
from supermetrics._generated.supermetrics_api_client.models.transfer_schedule import TransferSchedule
schedule = [TransferSchedule(run_interval="daily", run_hour=4)]
accounts = [TransferAccount(data_source_username="ads@example.com", login_id=1, account_id="8733197711")]
result = client.transfers.validate(
team_id=12345,
data_source_id="AW",
schema_id=99999,
destination_id=8,
display_name="Google Ads to BigQuery",
schedule=schedule,
accounts=accounts,
)
if not result.is_valid:
for error in result.errors:
print(f"{error.field_id}: {error.error_code}")
# Pause and resume. The API's vocabulary is "pause" / "unpause".
client.transfers.set_state(team_id=12345, transfer_id=36091, state="pause")
# Run history for a transfer, and the detail of one run
runs = client.transfers.list_runs(
team_id=12345,
transfer_id=36091,
start_date=datetime(2026, 1, 1, tzinfo=UTC),
end_date=datetime(2026, 1, 31, tzinfo=UTC),
)
for run in runs:
print(f"Run {run.id}: {run.status} ({run.total_rows} rows)")
run = client.transfer_runs.get(team_id=12345, transfer_run_id=12345)
for query in run.query_details:
print(f" query: {query.status}, {query.rows} rows")Destinations are the warehouses and buckets transfers write into. They live on the same Data Warehouse host as transfers, and the same client reaches them.
from supermetrics import SupermetricsClient
client = SupermetricsClient(api_key="your_api_key")
# What the team already has
for destination in client.destinations.list(team_id=12345):
print(f"{destination.id}: {destination.display_name} ({destination.type_})")
# Credentials are a plain dict — there is no request model to import, because the
# generated *Fields classes cannot be constructed. Keys depend on the type.
fields = {
"hostname": "any-domain.my-region.snowflakecomputing.com",
"warehouse": "DEMO_WH",
"database_name": "TEST_DB",
"schema": "PUBLIC",
"role": "ACCOUNTADMIN",
"username": "USER",
"private_key": "not-a-real-key",
}
# Try the credentials before committing to them. A connection that does not work is a
# returned result with success=False, not an exception.
result = client.destinations.test_connection(
team_id=12345,
type="DWH_SNOWFLAKE",
display_name="Snowflake (prod)",
fields=fields,
auth_method="AUTH_METHOD_KEY_PAIR",
)
if not result.success:
raise RuntimeError(f"Connection failed: {result.error}")
destination = client.destinations.create(
team_id=12345,
type="DWH_SNOWFLAKE",
display_name="Snowflake (prod)",
fields=fields,
auth_method="AUTH_METHOD_KEY_PAIR",
)
print(f"Created destination {destination.id}")
# get() answers with edit_settings, a list of UI form descriptors — not the flat fields
# dict create() and update() take. The read shape and the write shape differ.
stored = client.destinations.get(team_id=12345, destination_id=8)
for setting in stored.edit_settings:
print(f" {setting.id} ({setting.input_type}): {setting.value}")
# Check what still depends on a destination before removing it
usage = client.destinations.get_usage(team_id=12345, destination_id=8)
if usage.is_used:
for transfer in usage.transfers:
print(f"still used by {transfer.transfer_id}: {transfer.transfer_name}")
else:
client.destinations.delete(team_id=12345, destination_id=8)
# Rotate secrets for multiple destinations of the same type in one call
from supermetrics import BatchUpdateDestinationsBodyUpdatesItem
results = client.destinations.batch_update(
team_id=12345,
destination_type="DWH_SNOWFLAKE",
updates=[
BatchUpdateDestinationsBodyUpdatesItem(destination_id=8, new_secret="not-a-real-new-password"),
BatchUpdateDestinationsBodyUpdatesItem(destination_id=9, new_secret="not-a-real-new-password"),
],
)
if results.has_errors:
for item in results.results:
if item.status == "error":
print(f" destination {item.destination_id} failed: {item.error_code}")Table groups define the schema structure for data in the warehouse — tables, fields,
and their mappings. The team identity comes from the API key (no team_id parameter).
from supermetrics import (
EditTableGroupBody,
FieldDefinition,
ImportTableGroupBody,
SupermetricsClient,
TableDefinition,
TableGroupImport,
)
client = SupermetricsClient(api_key="your_api_key")
# List all table groups
for group in client.table_groups.list():
print(f"{group.group_id} (schema {group.schema_id}): {group.name}")
# Export a table group's full data model
export = client.table_groups.export(group_id="tg_123", version=1)
for table in export.tables:
print(f" {table.table_name}: {table.fields}")
# Import (create) a new table group
created = client.table_groups.import_(
body=ImportTableGroupBody(
version=1,
group=TableGroupImport(group_name="My Group", ds_id="AW", table_prefix="MYG"),
tables=[TableDefinition(table_name="CAMPAIGNS", fields=["campaign_id", "date"])],
fields=[FieldDefinition(field_id="campaign_id", target_name="campaign_id")],
)
)
print(f"Created: {created.group_id}")
# Edit (full replace) — export, modify, put back
export = client.table_groups.export(group_id=created.group_id, version=1)
updated = client.table_groups.edit(
group_id=created.group_id,
body=EditTableGroupBody(version=1, group=export.group, tables=export.tables, fields=export.fields),
)from supermetrics import SupermetricsClient
# Initialize client
client = SupermetricsClient(api_key="your_api_key")
# Create a backfill for historical data
backfill = client.backfills.create(team_id=12345, transfer_id=456789, range_start="2024-01-01", range_end="2024-01-31")
print(f"Backfill created: {backfill.transfer_backfill_id}")
print(f"Status: {backfill.status}")
# Get the latest backfill for a transfer
latest = client.backfills.get_latest(team_id=12345, transfer_id=456789)
print(f"Latest backfill status: {latest.status}")
print(f"Progress: {latest.transfer_runs_completed}/{latest.transfer_runs_total}")
# List all incomplete backfills for a team
backfills = client.backfills.list_incomplete(team_id=12345)
for bf in backfills:
print(f"Backfill {bf.transfer_backfill_id}: {bf.status}")
# Cancel a backfill
cancelled = client.backfills.cancel(team_id=12345, backfill_id=67890)
print(f"Backfill cancelled: {cancelled.status}")The client accepts exactly one of api_key, bearer_token, or token_provider:
from supermetrics import SupermetricsAsyncClient, SupermetricsClient
# Static API key
client = SupermetricsClient(api_key="api_live_abc123")
# OAuth access token
client = SupermetricsClient(bearer_token="otok_abc123")
# Dynamic provider, re-evaluated on every request so short-lived tokens can be
# refreshed without discarding the connection pool
async def get_valid_token() -> str:
return await oauth_service.get_access_token(team_id=123)
client = SupermetricsAsyncClient(token_provider=get_valid_token)Every resource method takes per-request auth_token, headers, and timeout overrides,
so one shared client can serve concurrent callers that each bring their own credential and
tracing context:
sync_client = SupermetricsClient(api_key="api_live_abc123")
login = sync_client.logins.get(
"login_abc123",
auth_token="otok_this_caller",
headers={"X-Span-Id": "a8f3b2c9", "Idempotency-Key": "req-42"},
timeout=120.0,
)Use with_raw_response when you need the HTTP status, headers, or raw payload alongside
the parsed model:
response = sync_client.with_raw_response.logins.get("login_abc123")
print(response.status_code, response.span_id, response.retry_after)
print(response.data.username)See Authentication & Transport for the full guide.
See the examples/ directory for complete working examples:
complete_flow.py- Full sync workflow from authentication to query executionasync_flow.py- Async version of complete workflowconnector_builder_flow.py- Connector Builder end-to-end operations (supports--base-urlfor local dev)
See examples/README.md for setup and running instructions.
The SDK provides specific exception types for different error scenarios:
from supermetrics import (
APIError,
NetworkError,
SupermetricsAuthError,
SupermetricsNotFoundError,
SupermetricsRateLimitError,
SupermetricsValidationError,
SupermetricsClient,
)
client = SupermetricsClient(api_key="your_key")
try:
link = client.login_links.create(ds_id="GAWA", description="Test")
except SupermetricsAuthError as e:
# e.error_code carries the upstream OAuth code, e.g. ACCESS_TOKEN_INVALID
print(f"Credential rejected ({e.error_code}): {e.message}")
except SupermetricsValidationError as e:
print(f"Invalid parameters: {e.message}")
except SupermetricsNotFoundError as e:
print(f"Not found: {e.message}")
except SupermetricsRateLimitError as e:
print(f"Throttled; retry after {e.retry_after}s")
except APIError as e:
# Any other HTTP error. Carries status_code, headers, error_code and details.
print(f"API error {e.status_code}: {e.message}")
except NetworkError as e:
print(f"Network error: {e.message}")AuthenticationError and ValidationError remain available as aliases of
SupermetricsAuthError and SupermetricsValidationError, and every HTTP error is a
subclass of APIError.
- Authentication & Transport - Credentials, per-request overrides, response metadata, error taxonomy
- Examples - Working code examples
- Scripts - OpenAPI filtering, patching, and SDK generation
The SDK client is auto-generated from the Supermetrics OpenAPI specification.
- Location:
openapi-specs/directory (containsopenapi-data.yaml,openapi-managment.yaml,openapi-team.yaml,openapi-connector-builder.yaml) - Merged Spec:
openapi-spec.yaml(project root) - filtered, patched, and merged from source specs - Configuration:
scripts/references/sdk-endpoint-filters.yaml- controls which endpoints are included and applies patches/customizations - Documentation: See scripts/README.md for detailed patch system documentation
The SDK uses a configuration-driven process to create a focused, customizable client from multiple OpenAPI specifications.
This YAML file defines which API endpoints to include in the SDK and allows you to apply patches/customizations to both endpoints and shared components.
Key Features:
- Endpoint Filtering: Include only the endpoints your application needs
- Endpoint Patches: Customize individual endpoint definitions (descriptions, parameters, responses, etc.)
- Component Patches: Apply surgical modifications to shared schemas, responses, and other components
- Merge & Replace Strategies: Deep merge or complete replacement of OpenAPI sections
Basic Example:
endpoints:
- method: GET
path: /ds/logins
- method: GET
path: /query/data/json
component_patches:
schemas:
DataResponse:
merge:
properties:
meta:
properties:
result:
properties:
cache_time:
nullable: trueFor detailed documentation on the configuration format, patch strategies, and comprehensive examples, see scripts/README.md.
This Python script processes multiple OpenAPI specifications, applies customizations, and creates a single openapi-spec.yaml file.
What it does:
- Reads configuration from
scripts/references/sdk-endpoint-filters.yaml - Scans and loads all
.yaml/.ymlfiles fromopenapi-specs/directory - Filters endpoints based on configuration
- Applies endpoint patches (merge/replace operations)
- Collects all referenced components via
$reftraversal (dependency resolution) - Resolves external file references
- Applies component patches to shared schemas, responses, etc.
- Detects and fails on duplicate
METHOD|PATHacross specs - Merges everything into single specification
- Validates all requested endpoints were found
Usage:
python scripts/filter_openapi_spec.pyConfiguration:
- Input:
openapi-specs/*.yamlandscripts/references/sdk-endpoint-filters.yaml - Output:
openapi-spec.yaml
Exit codes:
0- Success1- Error (missing files, duplicates, or validation failure)
For detailed documentation on patch strategies, troubleshooting, and examples, see scripts/README.md.
Full Regeneration (recommended):
# 1. Update source specs in openapi-specs/ if needed
# 2. Update scripts/references/sdk-endpoint-filters.yaml to add/remove endpoints or apply patches
# 3. Run filter script to regenerate merged spec
python scripts/filter_openapi_spec.py
# 4. Regenerate SDK from merged spec
./scripts/regenerate_client.shQuick Regeneration (if openapi-spec.yaml unchanged):
./scripts/regenerate_client.shregenerate_client.sh generates into a staging directory and replaces
src/supermetrics/_generated/ only once generation has succeeded, so a failed run leaves
the committed client untouched. It runs the generator through uvx on a pinned Python
3.12 — openapi-python-client cannot run on this project's default 3.14 interpreter —
at the version read out of pyproject.toml. Set GENERATOR_PYTHON to override. See
docs/openapi-generation.md.
- Monthly (or when Supermetrics API changes)
- After updating source specs in
openapi-specs/ - After modifying
scripts/references/sdk-endpoint-filters.yaml(adding/removing endpoints or changing patches)
- Edit
scripts/references/sdk-endpoint-filters.yaml:- Add/remove endpoints in the
endpointslist - Add/modify patches in
component_patchesor endpoint-levelpatches
- Add/remove endpoints in the
- Run
python scripts/filter_openapi_spec.pyto regenerate the merged spec - Run
./scripts/regenerate_client.shto regenerate the SDK client
See scripts/README.md for detailed documentation on:
- Configuration file format
- Endpoint and component patch strategies
- Comprehensive examples
- Troubleshooting guide
Note: The adapter pattern (implemented in Story 1.3+) protects users from breaking changes during regeneration
Contributions are welcome! Please see CONTRIBUTING.md for guidelines on how to contribute, run tests, and deploy releases.
Note: Every pull request must include an update to HISTORY.md describing the change under the relevant version section.
This package was created with Cookiecutter and the audreyfeldroy/cookiecutter-pypackage project template.