如何在Snowflake中配置并调用Workday API（含JWT生成）

一、前置准备 #

• 从Workday获取客户端ID、租户ID，以及用于签名JWT的RSA私钥（PEM格式）
• 确保Snowflake账号拥有以下权限：CREATE EXTERNAL ACCESS INTEGRATION（用于外部API访问）、CREATE PROCEDURE/CREATE FUNCTION（实现JWT生成逻辑）、对应仓库和数据库的USAGE权限

二、在Snowflake中生成JWT #

Snowflake无内置JWT生成函数，需通过Python UDF或存储过程实现RSA签名逻辑：

方案1：Python UDF生成JWT #

1. 先创建安全对象存储私钥（避免明文泄露）：

CREATE OR REPLACE SECRET workday_private_key
  TYPE = RSA_PRIVATE_KEY
  SECRET_STRING = '-----BEGIN RSA PRIVATE KEY-----
... 你的私钥内容 ...
-----END RSA PRIVATE KEY-----';

2. 创建Python UDF调用pyjwt库生成符合Workday标准的JWT：

CREATE OR REPLACE FUNCTION generate_workday_jwt(client_id STRING, tenant_id STRING)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
HANDLER = 'generate_jwt'
SECRETS = ('private_key' = workday_private_key)
AS $$
import jwt
import time
from cryptography.hazmat.primitives import serialization

def generate_jwt(client_id, tenant_id):
    private_key_pem = secrets.private_key
    private_key = serialization.load_pem_private_key(
        private_key_pem.encode(),
        password=None,
    )
    # Workday要求的JWT标准Claims
    payload = {
        "iss": client_id,
        "sub": client_id,
        "aud": f"https://{tenant_id}.workday.com",
        "exp": int(time.time()) + 3600,  # 有效期设1小时，不超过Workday建议的24小时上限
        "iat": int(time.time())
    }
    # 用RS256算法签名生成JWT
    jwt_token = jwt.encode(payload, private_key, algorithm="RS256")
    return jwt_token
$$;

3. 调用UDF生成JWT：

SELECT generate_workday_jwt('你的客户端ID', '你的租户ID') AS workday_jwt;

方案2：存储过程封装生成逻辑（带缓存） #

如果需要重复使用JWT避免频繁生成，可通过存储过程加缓存逻辑：

CREATE OR REPLACE PROCEDURE get_workday_jwt(client_id STRING, tenant_id STRING)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
HANDLER = 'get_jwt'
SECRETS = ('private_key' = workday_private_key)
AS $$
import jwt
import time
from cryptography.hazmat.primitives import serialization

# 简单缓存：存储上次生成的JWT和过期时间
cached_token = None
cached_expiry = 0

def get_jwt(client_id, tenant_id):
    global cached_token, cached_expiry
    current_time = int(time.time())
    # 检查缓存是否有效
    if cached_token and current_time < cached_expiry - 60:
        return cached_token
    
    # 重新生成JWT
    private_key_pem = secrets.private_key
    private_key = serialization.load_pem_private_key(
        private_key_pem.encode(),
        password=None,
    )
    payload = {
        "iss": client_id,
        "sub": client_id,
        "aud": f"https://{tenant_id}.workday.com",
        "exp": current_time + 3600,
        "iat": current_time
    }
    cached_token = jwt.encode(payload, private_key, algorithm="RS256")
    cached_expiry = current_time + 3600
    return cached_token
$$;

调用存储过程：

CALL get_workday_jwt('你的客户端ID', '你的租户ID');

三、配置Snowflake外部访问集成 #

要让Snowflake发起外部API请求，需创建网络规则和外部访问集成：

-- 允许访问Workday API域名的网络规则
CREATE OR REPLACE NETWORK RULE workday_network_rule
  TYPE = HOST_PORT
  VALUE_LIST = ('<你的租户ID>.workday.com:443')
  MODE = EGRESS;

-- 绑定网络规则的外部访问集成
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION workday_api_access
  ALLOWED_NETWORK_RULES = ('workday_network_rule')
  ENABLED = TRUE;

四、调用Workday API #

方式1：SQL原生REQUEST函数（适合简单GET请求） #

SET jwt_token = (SELECT generate_workday_jwt('你的客户端ID', '你的租户ID'));

SELECT REQUEST(
    'GET',
    'https://<你的租户ID>.workday.com/ccx/api/v1/your-target-endpoint',
    {'Authorization': 'Bearer ' || $jwt_token},
    '',
    'workday_api_access'
) AS api_response;

方式2：Python存储过程（适合复杂请求，如POST、带JSON payload） #

CREATE OR REPLACE PROCEDURE call_workday_api(client_id STRING, tenant_id STRING, api_endpoint STRING, request_body VARIANT)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
HANDLER = 'call_api'
SECRETS = ('private_key' = workday_private_key)
EXTERNAL_ACCESS_INTEGRATIONS = (workday_api_access)
PACKAGES = ('requests')
AS $$
import requests
import jwt
import time
from cryptography.hazmat.primitives import serialization

def call_api(client_id, tenant_id, api_endpoint, request_body):
    # 生成JWT
    private_key_pem = secrets.private_key
    private_key = serialization.load_pem_private_key(
        private_key_pem.encode(),
        password=None,
    )
    payload = {
        "iss": client_id,
        "sub": client_id,
        "aud": f"https://{tenant_id}.workday.com",
        "exp": int(time.time()) + 3600,
        "iat": int(time.time())
    }
    jwt_token = jwt.encode(payload, private_key, algorithm="RS256")
    
    # 发起API请求
    headers = {
        'Authorization': f'Bearer {jwt_token}',
        'Content-Type': 'application/json'
    }
    response = requests.post(
        f"https://{tenant_id}.workday.com{api_endpoint}",
        json=request_body,
        headers=headers
    )
    response.raise_for_status()  # 捕获HTTP错误
    return response.json()
$$;

调用存储过程发起POST请求：

CALL call_workday_api(
    '你的客户端ID',
    '你的租户ID',
    '/ccx/api/v1/your-post-endpoint',
    PARSE_JSON('{"key": "value"}')
);

五、关键注意事项 #

• JWT有效期：严格控制在Workday允许范围内，建议1小时，避免过期导致请求失败
• 私钥安全：必须通过Snowflake SECRET对象存储，禁止明文写入代码
• 权限管控：限制UDF/存储过程的调用权限，仅授权给需要的用户角色
• 错误处理：在存储过程中添加异常捕获逻辑，处理JWT生成失败、API超时、权限错误等场景
• 网络适配：若Snowflake部署在私有网络，需确保能访问Workday公网API，或配置Workday私有链接（如有支持）

内容的提问来源于stack exchange，提问作者Amit Malik