supermetrics-python-sdk

API Reference

Complete reference for the supermetrics Python SDK public API.

Client Classes

SupermetricsClient

Synchronous client for blocking I/O operations.

from supermetrics import SupermetricsClient

client = SupermetricsClient(
    api_key="your_api_key",
    user_agent=None,
    custom_headers=None,
    timeout=30.0,
    base_url="https://api.supermetrics.com"
)

Parameters:

Resource Properties:

Methods:

Example:

# Using context manager (recommended)
with SupermetricsClient(api_key="your_key") as client:
    accounts = client.accounts.list(ds_id="GAWA")

# Manual lifecycle
client = SupermetricsClient(api_key="your_key")
try:
    accounts = client.accounts.list(ds_id="GAWA")
finally:
    client.close()

SupermetricsAsyncClient

Asynchronous client for non-blocking I/O operations.

from supermetrics import SupermetricsAsyncClient

client = SupermetricsAsyncClient(
    api_key="your_api_key",
    user_agent=None,
    custom_headers=None,
    timeout=30.0,
    base_url="https://api.supermetrics.com"
)

Parameters: Same as SupermetricsClient

Resource Properties:

Methods:

Example:

import asyncio
from supermetrics import SupermetricsAsyncClient

async def main():
    async with SupermetricsAsyncClient(api_key="your_key") as client:
        accounts = await client.accounts.list(ds_id="GAWA")
        print(f"Found {len(accounts)} accounts")

asyncio.run(main())

Resources

LoginLinksResource

Manage data source authentication links.

create()

Create a login link for data source authentication.

link = client.login_links.create(
    ds_id="GAWA",
    description="My Analytics Connection",
    expiry_time=None,
    **kwargs
)

Parameters:

Returns: LoginLink object

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

link = client.login_links.create(
    ds_id="GAWA",
    description="Analytics Connection for Q1 Report"
)
print(f"Authentication URL: {link.login_url}")
print(f"Link ID: {link.link_id}")

get()

Retrieve a login link by ID.

link = client.login_links.get(link_id="abc123")

Parameters:

Returns: LoginLink object with current state

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

link = client.login_links.get(link_id="abc123")
if link.login_id:
    print(f"User authenticated successfully! Login ID: {link.login_id}")
else:
    print(f"Link status: {link.status_code}")

list()

List all login links for the authenticated user.

links = client.login_links.list()

Returns: List of LoginLink objects

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

links = client.login_links.list()
for link in links:
    print(f"{link.ds_name}: {link.status_code}")

close()

Close/expire a login link.

client.login_links.close(link_id="abc123")

Parameters:

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

client.login_links.close(link_id="abc123")
print("Link closed successfully")

LoginsResource

Retrieve login information and credentials.

get()

Retrieve a login by login ID.

login = client.logins.get(login_id="login_abc123")

Parameters:

Returns: DataSourceLogin object

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

login = client.logins.get(login_id="login_abc123")
print(f"Username: {login.username}")
print(f"Display Name: {login.display_name}")
print(f"Data Source: {login.ds_info.ds_name}")

list()

List all logins for the authenticated user.

logins = client.logins.list()

Returns: List of DataSourceLogin objects

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

logins = client.logins.list()
for login in logins:
    print(f"{login.ds_info.ds_name}: {login.username}")

get_by_username()

Retrieve a login by username (convenience method).

login = client.logins.get_by_username(login_username="user@example.com")

Parameters:

Returns: DataSourceLogin object

Raises: AuthenticationError, ValidationError, APIError, NetworkError, ValueError (if not found)

Example:

try:
    login = client.logins.get_by_username("analytics@company.com")
    print(f"Found login: {login.login_id}")
except ValueError:
    print("Login not found")

AccountsResource

Retrieve data source accounts available for querying.

list()

List all accounts for a data source.

accounts = client.accounts.list(
    ds_id="GAWA",
    login_usernames=None,
    cache_minutes=None
)

Parameters:

Returns: Flattened list of account objects with account_id, account_name, group_name

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

# List all GAWA accounts
accounts = client.accounts.list(ds_id="GAWA")

# Filter by specific username
accounts = client.accounts.list(
    ds_id="GAWA",
    login_usernames="analytics@company.com"
)

# Filter by multiple usernames
accounts = client.accounts.list(
    ds_id="google_ads",
    login_usernames=["user1@company.com", "user2@company.com"]
)

# Print account details
for account in accounts:
    print(f"{account.account_name} ({account.account_id})")

QueriesResource

Execute data queries to retrieve marketing data.

execute()

Execute a data query.

result = client.queries.execute(
    ds_id="GAWA",
    ds_accounts=["123456789"],
    fields=["Date", "Sessions", "Users"],
    start_date="2024-01-01",
    end_date="2024-01-07",
    **kwargs
)

Parameters:

Returns: DataResponse object or None

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

result = client.queries.execute(
    ds_id="GAWA",
    ds_accounts=["123456789"],
    fields=["Date", "Sessions", "Users", "Pageviews"],
    start_date="2024-01-01",
    end_date="2024-01-31",
    max_rows=10000,
    filter_="Sessions > 100"
)

if result and result.data:
    print(f"Retrieved {len(result.data)} rows")
    for row in result.data:
        print(row)

get_results()

Retrieve results for a previously executed async query.

result = client.queries.get_results(query_id="query_abc123")

Parameters:

Returns: DataResponse object or None

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

# Execute query
result = client.queries.execute(
    ds_id="GAWA",
    ds_accounts=["123456789"],
    fields=["Date", "Sessions"],
    start_date="2024-01-01",
    end_date="2024-12-31"
)

# Check if async processing
if result and result.meta and result.meta.status_code == "pending":
    print(f"Query is processing... Request ID: {result.meta.request_id}")

    # Poll for results
    import time
    time.sleep(5)
    result = client.queries.get_results(query_id=result.meta.request_id)

    if result and result.meta.status_code == "success":
        print(f"Query completed! Rows: {len(result.data)}")

BackfillsResource

Schedule and manage historical data backfills for Data Warehouse transfers.

Important: Backfill endpoints are served from a separate base URL. You must initialize the client with base_url="https://dts-api.supermetrics.com/v1" when using any backfill operations:

client = SupermetricsClient(
    api_key="your_api_key",
    base_url="https://dts-api.supermetrics.com/v1"
)

Required API key scopes:

Required user permissions:

create()

Schedule a new backfill for a transfer.

backfill = client.backfills.create(
    team_id=12345,
    transfer_id=456789,
    range_start=date(2024, 1, 1),
    range_end=date(2024, 1, 31)
)

Parameters:

Returns: Backfill object with status "CREATED"

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Notes:

Example:

from datetime import date
from supermetrics import SupermetricsClient

with SupermetricsClient(api_key="your_key") as client:
    backfill = client.backfills.create(
        team_id=12345,
        transfer_id=456789,
        range_start=date(2024, 1, 1),
        range_end=date(2024, 1, 31)
    )
    print(f"Backfill ID: {backfill.transfer_backfill_id}")
    print(f"Status: {backfill.status}")  # "CREATED"
    print(f"Total runs: {backfill.transfer_runs_total}")

get()

Retrieve a backfill by its ID.

backfill = client.backfills.get(team_id=12345, backfill_id=67890)

Parameters:

Returns: Backfill object with current status and progress

Raises: AuthenticationError, APIError (404 if not found), NetworkError

Example:

backfill = client.backfills.get(team_id=12345, backfill_id=67890)
print(f"Status: {backfill.status}")
print(f"Progress: {backfill.transfer_runs_completed}/{backfill.transfer_runs_total}")

if backfill.error_report:
    for err in backfill.error_report:
        print(f"Error on {err.transfer_run_date}: {err.error}")

get_latest()

Retrieve the most recent backfill for a transfer.

backfill = client.backfills.get_latest(team_id=12345, transfer_id=456789)

Parameters:

Returns: Backfill object — the latest backfill regardless of status

Raises: AuthenticationError, APIError (404 if no backfill has ever been created), NetworkError

Example:

try:
    latest = client.backfills.get_latest(team_id=12345, transfer_id=456789)
    print(f"Latest backfill: {latest.transfer_backfill_id}")
    print(f"Status: {latest.status}")
    print(f"Range: {latest.range_start_date}{latest.range_end_date}")
except APIError as e:
    if e.status_code == 404:
        print("No backfill has been created for this transfer yet")

list_incomplete()

List all incomplete backfills for a team.

backfills = client.backfills.list_incomplete(team_id=12345)

Parameters:

Returns: list[Backfill] — backfills with status CREATED, SCHEDULED, RUNNING, or FAILED, sorted by creation time (newest first). Returns an empty list if none exist.

Raises: AuthenticationError, APIError, NetworkError

Example:

backfills = client.backfills.list_incomplete(team_id=12345)

if not backfills:
    print("No incomplete backfills")
else:
    for backfill in backfills:
        print(f"[{backfill.status}] Backfill {backfill.transfer_backfill_id} "
              f"(transfer {backfill.transfer_id}): "
              f"{backfill.transfer_runs_completed}/{backfill.transfer_runs_total} runs done")

cancel()

Cancel a backfill by setting its status to "CANCELLED".

backfill = client.backfills.cancel(team_id=12345, backfill_id=67890)

Parameters:

Returns: Backfill object with status "CANCELLED" and updated timestamps

Raises: AuthenticationError, ValidationError (if backfill is already in a final state), APIError (404 if not found), NetworkError

Notes:

Example:

from supermetrics import ValidationError, APIError

try:
    cancelled = client.backfills.cancel(team_id=12345, backfill_id=67890)
    print(f"Status: {cancelled.status}")        # "CANCELLED"
    print(f"Ended at: {cancelled.end_time}")
except ValidationError:
    print("Cannot cancel — backfill is already in a final state")
except APIError as e:
    if e.status_code == 404:
        print("Backfill not found")

Async usage (all methods above are also available on BackfillsAsyncResource):

import asyncio
from datetime import date
from supermetrics import SupermetricsAsyncClient

async def main():
    async with SupermetricsAsyncClient(
        api_key="your_key",
        base_url="https://dts-api.supermetrics.com/v1"
    ) as client:
        backfill = await client.backfills.create(
            team_id=12345,
            transfer_id=456789,
            range_start=date(2024, 1, 1),
            range_end=date(2024, 1, 31)
        )
        print(f"Created backfill: {backfill.transfer_backfill_id}")

        incomplete = await client.backfills.list_incomplete(team_id=12345)
        print(f"Incomplete backfills: {len(incomplete)}")

asyncio.run(main())

DatasourceDetailsResource

Retrieve complete configuration details for a Supermetrics data source.

get()

Fetch metadata for a data source including report types, settings, and authentication requirements.

details = client.datasource_details.get(
    team_id=12345,
    data_source_id="GAWA",
    sm_app_id=None   # optional
)

Parameters:

Returns: DatasourceDetails object

Raises: AuthenticationError, ValidationError, APIError, NetworkError

Example:

from supermetrics import SupermetricsClient, APIError

with SupermetricsClient(api_key="your_key") as client:
    details = client.datasource_details.get(team_id=12345, data_source_id="GAWA")

    print(f"Name:    {details.name}")
    print(f"Status:  {details.status}")
    print(f"Premium: {details.is_premium}")

    if details.report_types:
        for rt in details.report_types:
            print(f"  Report type: {rt.id}{rt.name}")

Async usage:

import asyncio
from supermetrics import SupermetricsAsyncClient

async def main():
    async with SupermetricsAsyncClient(api_key="your_key") as client:
        details = await client.datasource_details.get(
            team_id=12345,
            data_source_id="GAWA"
        )
        print(f"Datasource: {details.name} ({details.status})")

asyncio.run(main())

Models

Backfill

Represents a Data Warehouse backfill job.

Attributes:

Example:

backfill = client.backfills.get(team_id=12345, backfill_id=67890)

# Check status
print(f"Status: {backfill.status}")

# Track progress
total = backfill.transfer_runs_total
done = backfill.transfer_runs_completed
failed = backfill.transfer_runs_failed
print(f"Progress: {done}/{total} completed, {failed} failed")

# Check for errors
for err in backfill.error_report:
    print(f"  {err.transfer_run_date}: {err.error}")

TransferBackfillRunError

Represents an error that occurred during a single transfer run within a backfill.

Attributes:

Example:

backfill = client.backfills.get(team_id=12345, backfill_id=67890)
for err in backfill.error_report:
    print(f"Run on {err.transfer_run_date} failed: {err.error}")

DatasourceDetails

Represents complete configuration metadata for a Supermetrics data source.

Key Attributes:

Example:

details = client.datasource_details.get(team_id=12345, data_source_id="GAWA")

print(f"ID:      {details.id}")
print(f"Name:    {details.name}")
print(f"Status:  {details.status}")
print(f"Premium: {details.is_premium}")
print(f"Auth required: {details.is_authentication_required}")

for rt in (details.report_types or []):
    print(f"  Report type: {rt.id}{rt.name}")

Represents a data source login link for OAuth authentication.

Key Attributes:

Example:

link = client.login_links.create(ds_id="GAWA")
print(f"Visit: {link.login_url}")
print(f"Expires: {link.expiry_time}")

DataSourceLogin

Represents an authenticated data source login with credentials.

Key Attributes:

Example:

login = client.logins.get(login_id="login_abc123")
print(f"Authenticated as: {login.username}")
print(f"Data Source: {login.ds_info.ds_name}")
print(f"Expires: {login.expiry_time}")

Account

Represents a data source account.

Attributes:

Example:

accounts = client.accounts.list(ds_id="GAWA")
for account in accounts:
    print(f"ID: {account.account_id}")
    print(f"Name: {account.account_name}")
    print(f"Group: {account.group_name}")

DataResponse

Represents a query execution response.

Key Attributes:

Example:

result = client.queries.execute(...)

# Check metadata
if result and result.meta:
    print(f"Status: {result.meta.status_code}")
    print(f"Request ID: {result.meta.request_id}")

# Process data
if result and result.data:
    headers = [field["field_id"] for field in result.meta.fields]
    print(f"Columns: {headers}")

    for row in result.data:
        print(row)

Exceptions

All exceptions inherit from SupermetricsError base class.

SupermetricsError

Base exception for all SDK errors.

Attributes:

Example:

from supermetrics import SupermetricsError

try:
    client.accounts.list(ds_id="GAWA")
except SupermetricsError as e:
    print(f"Error: {e.message}")
    print(f"Status: {e.status_code}")
    print(f"Endpoint: {e.endpoint}")

AuthenticationError

Raised when API authentication fails (HTTP 401).

Common Causes:

Example:

from supermetrics import AuthenticationError

try:
    client = SupermetricsClient(api_key="invalid_key")
    client.login_links.list()
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")

ValidationError

Raised when request validation fails (HTTP 400, 422).

Common Causes:

Example:

from supermetrics import ValidationError

try:
    client.accounts.list(ds_id="")  # Invalid empty ds_id
except ValidationError as e:
    print(f"Validation failed: {e.message}")

APIError

Raised for API-level errors (HTTP 403, 404, 429, 5xx).

Common Causes:

Example:

from supermetrics import APIError

try:
    client.logins.get("nonexistent_id")
except APIError as e:
    if e.status_code == 404:
        print("Login not found")
    elif e.status_code == 429:
        print("Rate limited")

NetworkError

Raised for network-level failures.

Common Causes:

Example:

from supermetrics import NetworkError

try:
    client = SupermetricsClient(api_key="key", timeout=1.0)
    client.login_links.list()
except NetworkError as e:
    print(f"Network error: {e.message}")

Type Utilities

UNSET and Unset

Sentinel value to distinguish between None and unset optional fields.

Example:

from supermetrics._generated.supermetrics_api_client.types import UNSET

link = client.login_links.get(link_id="abc123")

# Check if field is set
if link.description is not UNSET:
    print(f"Description: {link.description}")
else:
    print("No description provided")