fromfunquery.redashimportRedashClient# 方式一：直接传入参数client=RedashClient( redash_url="https://redash.example.com", api_key="your-api-key" ) # 方式二：从 funsecret 读取配置# 需要配置 visable.middleware.redash.redash_url 和 visable.middleware.redash.api_keyclient=RedashClient() # 验证凭证ifclient.test_credentials(): print("认证成功")

# 获取缓存的查询结果（仅无参数 Query）result=client.get_cached_query_result(query_id=123) # 执行查询并等待结果result=client.run_query_and_wait( query_id=123, parameters={"date_param": "2024-01-01"}, # 参数化查询max_age=0, # 0 表示强制重新执行timeout=60# 超时时间（秒） ) # 获取查询结果数据data=result["query_result"]["data"] columns=data["columns"] rows=data["rows"]

# 导出为 JSONjson_result=client.query_result(query_result_id=456, fmt="json") # 导出为 CSVcsv_text=client.query_result(query_result_id=456, fmt="csv") withopen("result.csv", "w") asf: f.write(csv_text)

# 获取查询列表（分页）queries=client.queries(page=1, page_size=25) # 获取所有查询（自动分页）all_queries=client.paginate(client.queries, page_size=100) # 获取单个查询query=client.query(query_id=123) # 创建查询new_query=client.create_query({"name": "My Query", "query": "SELECT * FROM users", "data_source_id": 1 }) # 更新查询client.update_query(query_id=123, data={"name": "Updated Name"}) # 复制查询copied_query=client.duplicate_query(query_id=123, new_name="Query Copy") # 归档查询client.archive_query(query_id=123) # 获取定时执行的查询scheduled=list(client.scheduled_queries())

# 获取 Dashboard 列表dashboards=client.dashboards(page=1, page_size=25) # 获取单个 Dashboarddashboard=client.get_dashboard(dashboard_id_or_slug="my-dashboard") # 创建 Dashboardnew_dashboard=client.create_dashboard(name="My Dashboard") # 更新 Dashboardclient.update_dashboard(dashboard_id=123, properties={"tags": ["tag1", "tag2"]}) # 复制 Dashboard（包含所有 Widget）copied_dashboard=client.duplicate_dashboard( slug="source-dashboard", new_name="Dashboard Copy" ) # 创建 Widgetclient.create_widget( dashboard_id=123, visualization_id=456, text="Widget Title", options={"position":{"col": 0, "row": 0, "sizeX": 3, "sizeY": 3}} ) # 归档 Dashboardclient.archive_dashboard(dashboard_slug="my-dashboard")

# 获取告警列表alerts=client.alerts() # 获取单个告警alert=client.get_alert(alert_id=123) # 创建告警new_alert=client.create_alert( name="My Alert", query_id=123, options={"column": "count", "op": ">", "value": 100} ) # 更新告警client.update_alert(alert_id=123, name="Updated Alert", rearm=300)

# 获取数据源列表data_sources=client.get_data_sources() # 获取单个数据源data_source=client.get_data_source(data_source_id=1) # 创建数据源new_ds=client.create_data_source( name="My PostgreSQL", _type="pg", options={"host": "localhost", "port": 5432, "dbname": "mydb", "user": "user", "password": "pass" } )

# 获取用户列表users=client.users(page=1, page_size=25) # 获取禁用的用户disabled_users=client.users(only_disabled=True) # 禁用用户client.disable_user(user_id=123)

# 收藏查询client.create_favorite(resource_type="query", resource_id=123) # 收藏 Dashboardclient.create_favorite(resource_type="dashboard", resource_id=456) # 获取收藏的查询favorite_queries=client.queries(only_favorites=True) # 获取收藏的 Dashboardfavorite_dashboards=client.dashboards(only_favorites=True)

importrequeststry: result=client.run_query_and_wait(query_id=123, timeout=30) exceptTimeoutErrorase: print(f"查询超时: {e}") exceptRuntimeErrorase: print(f"查询失败: {e}") exceptrequests.HTTPErrorase: print(f"HTTP 错误: {e}")