## Estimate assets for pricing before onboarding

URL: https://docs.atlan.com/get-started/references/asset-estimation

> Estimate asset counts for pricing inputs using SQL and API scripts for common platforms.

# Estimate assets for pricing before onboarding

Asset estimation determines the count of metadata objects across your data platforms. These counts are used for pricing inputs and implementation planning.

This reference helps you count metadata assets across your data platforms before onboarding.

## Snowflake

Use this query to estimate assets from `snowflake.account_usage`. 
It returns counts for tables, views, materialized views, external tables, columns, databases, and schemas.

```sql
WITH all_assets AS (
 SELECT
 table_catalog AS db_name,
 table_schema AS schema_name,
 table_name AS object_name,
 table_type AS asset_type
 FROM snowflake.account_usage.tables
 WHERE table_schema NOT LIKE 'INFORMATION_SCHEMA'
 AND table_type IN ('BASE TABLE', 'VIEW', 'MATERIALIZED VIEW', 'EXTERNAL TABLE')
 AND deleted IS NULL
 UNION ALL
 SELECT
 table_catalog,
 table_schema,
 table_name,
 'VIEW'
 FROM snowflake.account_usage.views
 WHERE table_schema NOT LIKE 'INFORMATION_SCHEMA'
 AND deleted IS NULL
),
column_counts AS (
 SELECT COUNT(*) AS column_count
 FROM snowflake.account_usage.columns
 WHERE deleted IS NULL
),
database_counts AS (
 SELECT COUNT(*) AS database_count
 FROM snowflake.account_usage.databases
 WHERE deleted IS NULL
),
schema_counts AS (
 SELECT COUNT(*) AS schema_count
 FROM snowflake.account_usage.schemata
 WHERE deleted IS NULL
),
asset_counts AS (
 SELECT asset_type, COUNT(*) AS count
 FROM all_assets
 GROUP BY asset_type
 UNION ALL
 SELECT 'TOTAL' AS asset_type, COUNT(*) AS count
 FROM all_assets
)
SELECT asset_type, count
FROM asset_counts
UNION ALL
SELECT 'COLUMN_COUNT', column_count FROM column_counts
UNION ALL
SELECT 'DATABASE_COUNT', database_count FROM database_counts
UNION ALL
SELECT 'SCHEMA_COUNT', schema_count FROM schema_counts;
```

## Databricks data warehouse

Use this query to estimate assets from `system.information_schema`. 
It returns counts for catalogs, schemas, tables, views, columns, volumes, and external locations.

```sql
/* 1. Database (Catalog) Count */
SELECT
 'Database (Catalog)' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.catalogs
WHERE
 catalog_name != 'system'
UNION ALL
/* 2. Schema Count */
SELECT
 'Schema' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.schemata
WHERE
 schema_name != 'information_schema'
UNION ALL
/* 3. Table Count (Managed & External) */
SELECT
 'Table' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.tables
WHERE
 table_type IN ('MANAGED', 'EXTERNAL')
UNION ALL
/* 4. Standard View Count */
SELECT
 'View' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.tables
WHERE
 table_type = 'VIEW'
UNION ALL
/* 5. Materialized View Count */
SELECT
 'Materialized View' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.tables
WHERE
 table_type = 'MATERIALIZED_VIEW'
UNION ALL
/* 6. Streaming Tables */
SELECT
 'Streaming Tables' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.tables
WHERE
 table_type = 'STREAMING_TABLE'
UNION ALL
/* 7. Column Count */
SELECT
 'Column' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.columns
UNION ALL
/* 8. Volume Count */
SELECT
 'Volume' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.volumes
UNION ALL
/* 9. External Location Count */
SELECT
 'External Location' AS asset_type,
 COUNT(*) AS count
FROM
 system.information_schema.external_locations;
```

## BigQuery

Column counts are scoped by region. 
Run this query for every region where datasets live, then sum the results.

```sql
SELECT COUNT(*) AS column_count
FROM `region-us`.INFORMATION_SCHEMA.COLUMNS
WHERE table_catalog = 'acme-analytics';
```

## Redshift

`information_schema` is scoped to a single database. 
Run this query for every database with tables or external tables, then sum the results.

```sql
SELECT COUNT(*) AS column_count
FROM SVV_COLUMNS
WHERE table_schema NOT IN ('pg_internal','pg_catalog','information_schema');
```

## SQL server

Run this query for every database you plan to crawl and total the results.

```sql
SELECT
 'Schema' AS asset_type,
 COUNT(*) AS count
FROM
 INFORMATION_SCHEMA.SCHEMATA
UNION ALL
SELECT
 'Table' AS asset_type,
 COUNT(*) AS count
FROM
 INFORMATION_SCHEMA.TABLES
WHERE
 TABLE_TYPE = 'BASE TABLE'
UNION ALL
SELECT
 'View' AS asset_type,
 COUNT(*) AS count
FROM
 INFORMATION_SCHEMA.TABLES
WHERE
 TABLE_TYPE = 'VIEW'
UNION ALL
SELECT
 'Column' AS asset_type,
 COUNT(*) AS count
FROM
 INFORMATION_SCHEMA.COLUMNS;
```

If stored procedures are in scope, run this query per database and add the total:

```sql
SELECT
 'Stored Procedure' AS asset_type,
 COUNT(*) AS count
FROM
 INFORMATION_SCHEMA.ROUTINES
WHERE
 ROUTINE_TYPE = 'PROCEDURE';
```

## Power BI

Use this script to estimate tenant assets with the Power BI Admin API. 
It uses the `Tenant.Read.All` permission and can reuse the same app registration as metadata crawling.

```python

# Configuration

# Option 1: Use existing access token (if you have one)

ACCESS_TOKEN = ""

# Option 2: Generate token automatically using Azure AD App credentials

# (Leave ACCESS_TOKEN empty to use this method)

TENANT_ID = "0f9b7b1e-6b70-4f3d-8c9f-3c81a1a3a9e2"
CLIENT_ID = "3b5d64f4-4f0c-4d38-9f62-b1f1a65f3db9"
CLIENT_SECRET = "pbi-client-secret-value"

BASE_URL = "https://api.powerbi.com/v1.0/myorg/admin"

def get_access_token():
 """Gets access token via client credentials flow or uses provided token."""
 if ACCESS_TOKEN:
 return ACCESS_TOKEN
 if not all([TENANT_ID, CLIENT_ID, CLIENT_SECRET]):
 raise ValueError("Provide ACCESS_TOKEN or TENANT_ID + CLIENT_ID + CLIENT_SECRET")
 url = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"
 data = {
 "grant_type": "client_credentials",
 "client_id": CLIENT_ID,
 "client_secret": CLIENT_SECRET,
 "scope": "https://analysis.windows.net/powerbi/api/.default"
 }
 response = requests.post(url, data=data)
 response.raise_for_status()
 return response.json()["access_token"]

def api_request(method, url, headers, **kwargs):
 """Makes API request with retry logic for rate limiting."""
 for attempt in range(5):
 response = requests.request(method, url, headers=headers, **kwargs)
 if response.status_code == 429:
 wait = int(response.headers.get("Retry-After", 60))
 print(f"Rate limited. Waiting {wait}s...")
 time.sleep(wait)
 continue
 response.raise_for_status()
 return response
 raise Exception("Max retries exceeded")

def get_all_workspaces(headers):
 """Retrieves all workspace IDs in the tenant with pagination."""
 url = f"{BASE_URL}/workspaces/modified"
 workspace_ids = []
 while url:
 response = api_request("GET", url, headers)
 data = response.json()
 workspace_ids.extend([w['id'] for w in data])
 url = response.headers.get("x-ms-continuation-url") # Handle pagination
 return workspace_ids

def trigger_scan(headers, workspace_ids):
 """Triggers async scan for a batch of workspaces."""
 url = f"{BASE_URL}/workspaces/getInfo?datasetSchema=True&datasetExpressions=True"
 response = api_request("POST", url, headers, json={"workspaces": workspace_ids})
 return response.json()['id']

def get_scan_result(headers, scan_id):
 """Polls for scan completion and returns the result."""
 status_url = f"{BASE_URL}/workspaces/scanStatus/{scan_id}"
 result_url = f"{BASE_URL}/workspaces/scanResult/{scan_id}"
 while True:
 response = api_request("GET", status_url, headers)
 status = response.json().get('status')
 if status == "Succeeded":
 print(f"Scan {scan_id} complete.")
 return api_request("GET", result_url, headers).json()
 if status == "Failed":
 raise Exception(f"Scan failed: {response.json()}")
 print(f"Scanning... Status: {status}")
 time.sleep(3)

def count_assets(scan_results):
 """Parses scan results to count all assets."""
 counts = {
 "Workspaces": 0,
 "Reports": 0,
 "Dashboards": 0,
 "Semantic Models (Datasets)": 0,
 "Tables": 0,
 "Fields (Columns + Measures)": 0,
 "Dataflows": 0,
 "Metrics (Goals)": 0
 }
 workspaces = scan_results.get('workspaces', [])
 counts["Workspaces"] = len(workspaces)
 for ws in workspaces:
 counts["Reports"] += len(ws.get('reports', []))
 counts["Dashboards"] += len(ws.get('dashboards', []))
 counts["Dataflows"] += len(ws.get('dataflows', []))
 for scorecard in ws.get('scorecards', []):
 counts["Metrics (Goals)"] += len(scorecard.get('goals', []))
 for ds in ws.get('datasets', []):
 counts["Semantic Models (Datasets)"] += 1
 for table in ds.get('tables', []):
 counts["Tables"] += 1
 counts["Fields (Columns + Measures)"] += len(table.get('columns', []))
 counts["Fields (Columns + Measures)"] += len(table.get('measures', []))
 return counts

if __name__ == "__main__":
 try:
 token = get_access_token()
 headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
 print("Fetching workspace list...")
 all_ws_ids = get_all_workspaces(headers)
 print(f"Found {len(all_ws_ids)} workspaces.")
 totals = {
 "Workspaces": 0,
 "Reports": 0,
 "Dashboards": 0,
 "Semantic Models (Datasets)": 0,
 "Tables": 0,
 "Fields (Columns + Measures)": 0,
 "Dataflows": 0,
 "Metrics (Goals)": 0
 }
 for i in range(0, len(all_ws_ids), 100):
 chunk = all_ws_ids[i:i + 100]
 print(f"Processing workspaces {i + 1} to {i + len(chunk)}...")
 scan_id = trigger_scan(headers, chunk)
 result = get_scan_result(headers, scan_id)
 for key, value in count_assets(result).items():
 totals[key] += value
 print("\n" + "=" * 40)
 print("POWER BI TENANT INVENTORY")
 print("=" * 40)
 for key, value in totals.items():
 print(f"{key}: {value:,}")
 except Exception as e:
 print(f"Error: {e}")
```

## Tableau

Use this script to estimate Tableau assets with the REST and Metadata APIs. 
Use the same credentials used for metadata crawling.

```python

# Tableau Server or Cloud URL

# Example Tableau Cloud: https://prod-useast-a.online.tableau.com

# Example Tableau Server: https://tableau.acme.corp

TABLEAU_HOST = "https://prod-useast-a.online.tableau.com"
SITE_NAME = "" # Leave empty for default site, or specify site content URL

# API Version

API_VERSION = "3.21"

# Option 1: Use existing auth token (if you have one from a previous session)

AUTH_TOKEN = ""

# Option 2: Connected App (JWT) authentication

CONNECTED_APP_CLIENT_ID = "tableau-client-id"
CONNECTED_APP_SECRET_ID = "tableau-secret-id"
CONNECTED_APP_SECRET_VALUE = "tableau-secret-value"
CONNECTED_APP_USERNAME = "data.engineer@acme.com"

# Option 3: Personal Access Token authentication

PAT_NAME = "atlan-metadata-crawl"
PAT_SECRET = "tableau-pat-secret"

# Option 4: Username/Password authentication

USERNAME = "data.engineer@acme.com"
PASSWORD = "tableau-password"

METADATA_API_URL = f"{TABLEAU_HOST}/api/metadata/graphql"

def generate_jwt(client_id, secret_id, secret_value, username):
 """Generates a JWT for Connected App authentication."""
 try:
 import jwt
 except ImportError:
 raise ImportError("PyJWT library required for Connected App auth. Install with: pip install PyJWT")
 token = jwt.encode(
 {
 "iss": client_id,
 "exp": int(time.time()) + 600,
 "jti": str(uuid.uuid4()),
 "aud": "tableau",
 "sub": username,
 "scp": ["tableau:content:read", "tableau:insight_definitions_metrics:read"]
 },
 secret_value,
 algorithm="HS256",
 headers={
 "kid": secret_id,
 "iss": client_id
 }
 )
 return token

def get_auth_token():
 """Gets auth token via Connected App, PAT, username/password, or uses provided token."""
 if AUTH_TOKEN:
 return AUTH_TOKEN, None
 site_id = None
 auth_headers = {"Content-Type": "application/json", "Accept": "application/json"}
 if all([CONNECTED_APP_CLIENT_ID, CONNECTED_APP_SECRET_ID, CONNECTED_APP_SECRET_VALUE, CONNECTED_APP_USERNAME]):
 print("Using Connected App (JWT) authentication...")
 jwt_token = generate_jwt(
 CONNECTED_APP_CLIENT_ID,
 CONNECTED_APP_SECRET_ID,
 CONNECTED_APP_SECRET_VALUE,
 CONNECTED_APP_USERNAME
 )
 url = f"{TABLEAU_HOST}/api/{API_VERSION}/auth/signin"
 payload = {
 "credentials": {
 "jwt": jwt_token,
 "site": {"contentUrl": SITE_NAME}
 }
 }
 response = requests.post(url, json=payload, headers=auth_headers)
 response.raise_for_status()
 data = response.json()
 site_id = data["credentials"]["site"]["id"]
 return data["credentials"]["token"], site_id
 if PAT_NAME and PAT_SECRET:
 print("Using Personal Access Token authentication...")
 url = f"{TABLEAU_HOST}/api/{API_VERSION}/auth/signin"
 payload = {
 "credentials": {
 "personalAccessTokenName": PAT_NAME,
 "personalAccessTokenSecret": PAT_SECRET,
 "site": {"contentUrl": SITE_NAME}
 }
 }
 response = requests.post(url, json=payload, headers=auth_headers)
 response.raise_for_status()
 data = response.json()
 site_id = data["credentials"]["site"]["id"]
 return data["credentials"]["token"], site_id
 if USERNAME and PASSWORD:
 print("Using username/password authentication...")
 url = f"{TABLEAU_HOST}/api/{API_VERSION}/auth/signin"
 payload = {
 "credentials": {
 "name": USERNAME,
 "password": PASSWORD,
 "site": {"contentUrl": SITE_NAME}
 }
 }
 response = requests.post(url, json=payload, headers=auth_headers)
 response.raise_for_status()
 data = response.json()
 site_id = data["credentials"]["site"]["id"]
 return data["credentials"]["token"], site_id
 raise ValueError(
 "Provide one of:\n"
 " 1. AUTH_TOKEN (existing token)\n"
 " 2. CONNECTED_APP_CLIENT_ID + SECRET_ID + SECRET_VALUE + USERNAME\n"
 " 3. PAT_NAME + PAT_SECRET (Personal Access Token)\n"
 " 4. USERNAME + PASSWORD"
 )

def api_request(method, url, headers, **kwargs):
 """Makes API request with retry logic for rate limiting."""
 for attempt in range(5):
 response = requests.request(method, url, headers=headers, **kwargs)
 if response.status_code == 429:
 wait = int(response.headers.get("Retry-After", 60))
 print(f"Rate limited. Waiting {wait}s...")
 time.sleep(wait)
 continue
 response.raise_for_status()
 return response
 raise Exception("Max retries exceeded")

def get_asset_counts(headers):
 """Queries the Tableau Metadata API to get asset counts."""
 query = """
 {
 workbooksConnection {
 totalCount
 }
 dashboardsConnection {
 totalCount
 }
 sheetsConnection {
 totalCount
 }
 publishedDatasourcesConnection {
 totalCount
 }
 embeddedDatasourcesConnection {
 totalCount
 }
 databaseTablesConnection {
 totalCount
 }
 fieldsConnection {
 totalCount
 }
 databasesConnection {
 totalCount
 }
 tablesConnection {
 totalCount
 }
 columnsConnection {
 totalCount
 }
 calculatedFieldsConnection {
 totalCount
 }
 }
 """
 response = api_request("POST", METADATA_API_URL, headers, json={"query": query})
 data = response.json()
 if "errors" in data:
 raise Exception(f"GraphQL Errors: {json.dumps(data['errors'], indent=2)}")
 return data["data"]

def parse_counts(data):
 """Parses GraphQL response into a structured counts dictionary."""
 counts = {
 "Workbooks": data["workbooksConnection"]["totalCount"],
 "Dashboards": data["dashboardsConnection"]["totalCount"],
 "Sheets (Views)": data["sheetsConnection"]["totalCount"],
 "Published Datasources": data["publishedDatasourcesConnection"]["totalCount"],
 "Embedded Datasources": data["embeddedDatasourcesConnection"]["totalCount"],
 "Databases (Upstream)": data["databasesConnection"]["totalCount"],
 "Database Tables (Upstream)": data["databaseTablesConnection"]["totalCount"],
 "Tables (Logical)": data["tablesConnection"]["totalCount"],
 "Columns": data["columnsConnection"]["totalCount"],
 "Calculated Fields": data["calculatedFieldsConnection"]["totalCount"],
 "All Fields (Total)": data["fieldsConnection"]["totalCount"],
 }
 return counts

def sign_out(headers, site_id):
 """Signs out and invalidates the auth token."""
 try:
 url = f"{TABLEAU_HOST}/api/{API_VERSION}/auth/signout"
 requests.post(url, headers=headers)
 print("Signed out successfully.")
 except Exception:
 pass

if __name__ == "__main__":
 site_id = None
 token = None
 try:
 print("Authenticating with Tableau...")
 token, site_id = get_auth_token()
 print("Authentication successful!")
 headers = {
 "X-Tableau-Auth": token,
 "Content-Type": "application/json",
 "Accept": "application/json"
 }
 print(f"Querying Tableau Metadata API at {METADATA_API_URL}...")
 data = get_asset_counts(headers)
 counts = parse_counts(data)
 print("\n" + "=" * 45)
 print("TABLEAU ASSET INVENTORY")
 print("=" * 45)
 print("\n-- BI Content --")
 print(f"{'Workbooks:':<30} {counts['Workbooks']:,}")
 print(f"{'Dashboards:':<30} {counts['Dashboards']:,}")
 print(f"{'Sheets (Views):':<30} {counts['Sheets (Views)']:,}")
 print("\n-- Datasources --")
 print(f"{'Published Datasources:':<30} {counts['Published Datasources']:,}")
 print(f"{'Embedded Datasources:':<30} {counts['Embedded Datasources']:,}")
 print("\n-- Upstream (Physical) --")
 print(f"{'Databases:':<30} {counts['Databases (Upstream)']:,}")
 print(f"{'Database Tables:':<30} {counts['Database Tables (Upstream)']:,}")
 print("\n-- Logical Model --")
 print(f"{'Tables:':<30} {counts['Tables (Logical)']:,}")
 print(f"{'Columns:':<30} {counts['Columns']:,}")
 print(f"{'Calculated Fields:':<30} {counts['Calculated Fields']:,}")
 print(f"{'All Fields (Total):':<30} {counts['All Fields (Total)']:,}")
 print("\n" + "-" * 45)
 total_bi_assets = (
 counts["Workbooks"] +
 counts["Dashboards"] +
 counts["Sheets (Views)"]
 )
 total_datasources = (
 counts["Published Datasources"] +
 counts["Embedded Datasources"]
 )
 total_governance = (
 total_bi_assets +
 counts["Published Datasources"] +
 counts["All Fields (Total)"]
 )
 print(f"{'Total BI Content:':<30} {total_bi_assets:,}")
 print(f"{'Total Datasources:':<30} {total_datasources:,}")
 print(f"{'Est. Governance Assets:':<30} ~{total_governance:,}")
 print("=" * 45)
 except requests.exceptions.HTTPError as e:
 print(f"HTTP Error: {e}")
 if e.response is not None:
 print(f"Response: {e.response.text}")
 except Exception as e:
 print(f"Error: {e}")
 finally:
 if token and site_id and not AUTH_TOKEN:
 headers = {"X-Tableau-Auth": token}
 sign_out(headers, site_id)
```

---