如何通过Python结合Power BI REST API向DAX查询传参并优化脚本

一、实现向DAX查询传递用户参数 #

你的DAX查询中使用了@user作为参数占位符，以下是两种可靠的参数传递实现方案：

方案1：字符串替换快速实现 #

直接将DAX查询中的@user占位符替换为目标用户邮箱，适合简单场景：

# 定义目标用户邮箱
target_user = "user@example.com"  # 替换为实际用户邮箱

# 替换DAX中的参数占位符，注意添加单引号符合DAX语法
parameterized_dax = dax_query.replace("@user", f"'{target_user}'")

payload_data = {
    "queries": [
        {
            "query": parameterized_dax,
        }
    ],
    "serializerSettings": {
        "includeNulls": True,
    },
}

注意：如果用户邮箱包含单引号等特殊字符，需要额外转义（比如把'替换为''），避免DAX语法错误。

方案2：API原生参数传递（推荐） #

使用Power BI REST API的参数化查询机制，更安全规范，同时优化DAX语句可读性：

1. 修改dax_query.py中的DAX，显式定义参数并简化过滤逻辑：

dax_query = """
DEFINE
    VAR @user = ""
Evaluate 
SUMMARIZE( 
    'axp_gw_succes_faild', 
    'axp_gw_succes_faild'[Date], 
    'axp_gw_succes_faild'[api_publisher], 
    'axp_gw_succes_faild'[application_name], 
    'axp_gw_succes_faild'[api_name],
    "TotalCount", 
    CALCULATE (
        SUM ( axp_gw_succes_faild[Count] ),
        FILTER(
            'Publisher-UserDetails',
            'Publisher-UserDetails'[BAdashboard_authoritylevel] IN {"Publisher", "publisherTechLead", "publisherArchitect", "ProductOwner"}
            && 'Publisher-UserDetails'[User-Email] = @user
        ),
        GROUPBY (axp_gw_succes_faild,axp_gw_succes_faild[api_name],axp_gw_succes_faild[Date])
    )
)
"""

2. 在Python脚本中添加参数传递逻辑：

target_user = "user@example.com"

payload_data = {
    "queries": [
        {
            "query": dax_query,
            "parameters": [
                {
                    "name": "user",
                    "value": target_user
                }
            ]
        }
    ],
    "serializerSettings": {
        "includeNulls": True,
    },
}

二、Python脚本优化建议 #

1. 增强错误处理 #

捕获网络异常、HTTP错误和JSON解析失败，避免脚本意外崩溃：

import json
import requests
from requests.exceptions import RequestException, JSONDecodeError
from cred import username, password, client_id, client_secret, scope, token_endpoint, url
from dax_query import dax_query

def getToken():
    try:
        response = requests.post(token_endpoint, data=payload, timeout=10)
        response.raise_for_status()  # 抛出4xx/5xx状态码错误
        response_data = response.json()
        access_token = response_data.get("access_token")
        if access_token:
            return access_token
        print("获取Token失败：响应中无access_token字段")
        exit(1)
    except RequestException as e:
        print(f"Token请求失败：{str(e)}")
        exit(1)
    except JSONDecodeError:
        print("Token响应解析JSON失败")
        exit(1)

2. 缓存Access Token #

Token有效期通常为1小时，添加缓存避免重复请求：

import time

token_cache = {"access_token": None, "expires_at": 0}

def getToken():
    current_time = time.time()
    # 检查缓存是否有效，提前60秒刷新避免过期
    if token_cache["access_token"] and current_time < token_cache["expires_at"] - 60:
        return token_cache["access_token"]
    
    try:
        response = requests.post(token_endpoint, data=payload, timeout=10)
        response.raise_for_status()
        response_data = response.json()
        access_token = response_data.get("access_token")
        expires_in = response_data.get("expires_in", 3600)
        if access_token:
            token_cache["access_token"] = access_token
            token_cache["expires_at"] = current_time + expires_in
            return access_token
        print("获取Token失败：响应中无access_token字段")
        exit(1)
    except RequestException as e:
        print(f"Token请求失败：{str(e)}")
        exit(1)
    except JSONDecodeError:
        print("Token响应解析JSON失败")
        exit(1)

3. 配置与代码分离 #

使用环境变量存储敏感配置，避免硬编码：

# 创建.env文件
USERNAME=your_username
PASSWORD=your_password
CLIENT_ID=your_client_id
CLIENT_SECRET=your_client_secret
SCOPE=https://analysis.windows.net/powerbi/api
TOKEN_ENDPOINT=https://login.microsoftonline.com/common/oauth2/token
POWER_BI_URL=https://api.powerbi.com/v1.0/myorg/datasets/{dataset_id}/executeQueries

from dotenv import load_dotenv
import os

load_dotenv()

username = os.getenv("USERNAME")
password = os.getenv("PASSWORD")
client_id = os.getenv("CLIENT_ID")
client_secret = os.getenv("CLIENT_SECRET")
scope = os.getenv("SCOPE")
token_endpoint = os.getenv("TOKEN_ENDPOINT")
url = os.getenv("POWER_BI_URL")

4. 代码结构化封装 #

将查询逻辑封装为函数，提升复用性：

def execute_dax_query(user_email):
    access_token = getToken()
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    }
    # 替换参数或使用API参数化方案
    parameterized_dax = dax_query.replace("@user", f"'{user_email}'")
    payload_data = {
        "queries": [{"query": parameterized_dax}],
        "serializerSettings": {"includeNulls": True},
    }
    try:
        response = requests.post(url, headers=headers, json=payload_data, timeout=15)
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        print(f"DAX查询执行失败：{str(e)}")
        return None
    except JSONDecodeError:
        print("查询响应解析JSON失败")
        return None

# 调用示例
if __name__ == "__main__":
    result = execute_dax_query("user@example.com")
    if result:
        print(json.dumps(result, indent=2))
    else:
        print("查询执行失败")

5. 日志替代print #

使用标准日志库记录信息，便于调试和生产环境监控：

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# 替换print语句
logger.info("成功获取Access Token")
logger.error(f"Token请求失败：{str(e)}")

6. 安全优化 #

• 禁止在日志中输出密码、Token等敏感信息
• 生产环境避免使用密码授权模式，优先选择客户端凭证流或授权码流
• 使用密钥管理服务存储敏感配置，而非明文文件

内容的提问来源于stack exchange，提问作者Ramesh Madhubhashitha