第10章 数据可视化¶

import tableauserverclient as TSC

# 连接到 Tableau Server
tableau_auth = TSC.TableauAuth('username', 'password', site_id='site_name')
server = TSC.Server('https://your-server.com', use_server_version=True)

with server.auth.sign_in(tableau_auth):
    # 查询已有项目
    all_projects, _ = server.projects.get()
    project = next(p for p in all_projects if p.name == 'Sales Project')

    # 发布数据源（从 .tdsx 文件）
    ds_item = TSC.DatasourceItem(project.id, name='Sales Data Source')
    ds_item = server.datasources.publish(
        ds_item,
        'sales_data.tdsx',
        TSC.Server.PublishMode.Overwrite
    )
    print(f"Datasource published: {ds_item.id}")

    # 发布工作簿（从 .twbx 文件）
    wb_item = TSC.WorkbookItem(project.id, name='Sales Dashboard')
    wb_item = server.workbooks.publish(
        wb_item,
        'sales_dashboard.twbx',
        TSC.Server.PublishMode.Overwrite
    )
    print(f"Workbook published: {wb_item.id}")

    # 查询工作簿中的视图
    server.workbooks.populate_views(wb_item)
    for view in wb_item.views:
        print(f"View: {view.name}, URL: {view.content_url}")

import looker_sdk

# 初始化 Looker SDK（使用 looker.ini 配置文件）
sdk = looker_sdk.init40()  # 使用 API 4.0

# 创建查询
body = looker_sdk.models40.WriteQuery(
    model="ecommerce",
    view="order_items",
    fields=["products.category", "order_items.total_revenue"],
    filters={"order_items.created_date": "30 days"},
    sorts=["order_items.total_revenue desc"],
    limit="10"
)
query = sdk.create_query(body=body)

# 运行查询并获取结果
result = sdk.run_query(query_id=query.id, result_format="json")
print(result)

# 获取已有 Dashboard
dashboards = sdk.search_dashboards(title="Sales")
for dash in dashboards:
    print(f"Dashboard: {dash.title}, ID: {dash.id}")

# 获取 Look 列表
looks = sdk.all_looks()
for look in looks[:5]:
    print(f"Look: {look.title}, ID: {look.id}")

# Superset 示例（通过 REST API 调用）
import requests

SUPERSET_URL = "http://localhost:8088"

# 登录获取 Token
login_resp = requests.post(
    f"{SUPERSET_URL}/api/v1/security/login",
    json={"username": "admin", "password": "admin", "provider": "db"}
)
access_token = login_resp.json()["access_token"]
headers = {"Authorization": f"Bearer {access_token}"}

# 获取数据库连接列表
db_resp = requests.get(f"{SUPERSET_URL}/api/v1/database/", headers=headers)
for db in db_resp.json()["result"]:
    print(f"Database: {db['database_name']}, ID: {db['id']}")

# 执行 SQL 查询（通过 SQL Lab API）
sql_result = requests.post(
    f"{SUPERSET_URL}/api/v1/sqllab/execute/",
    headers=headers,
    json={
        "database_id": 1,
        "sql": "SELECT sale_date, SUM(amount) as total FROM sales GROUP BY sale_date",
        "schema": "public"
    }
)
print(sql_result.json())

# 获取图表列表
charts_resp = requests.get(f"{SUPERSET_URL}/api/v1/chart/", headers=headers)
for chart in charts_resp.json()["result"][:5]:  # 切片操作：[start:end:step]提取子序列
    print(f"Chart: {chart['slice_name']}, Type: {chart['viz_type']}")