# 环境变量设置
env_vars允许在UDF调用时动态注入环境变量到Actor进程，无需为不同环境变量配置创建冗余UDF定义。
#### 基本用法
```
# Dataset API — 传入 dict
ds = ds.map(
    fn="my_schema.data_loader",
    on=["url_col"],
    env_vars={"DATA_SOURCE": "obs://bucket/path", "REGION": "cn-north-4"},
)
# with_arguments — 传入 dict 或 JSON 字符串
result = my_udf(t.col_a).with_arguments(
    env_vars={"DATA_SOURCE": "obs://bucket/path"},
)
```
在UDF中通过os.environ访问：
```
def my_udf(url: str) -> str:
    import os
    data_source = os.environ.get("DATA_SOURCE", "")
    return load_from(url, base=data_source)
```
#### 类型校验
Python侧在with_arguments() 中执行严格类型校验：
| 输入                        | 行为                            |
|:---|:---|
| env_vars={"KEY": "VALUE"} | 正常，key/value 均为 str           |
| env_vars={1: "val"}       | TypeError: keys must be str   |
| env_vars={"key": 123}     | TypeError: values must be str |
| env_vars={}               | 等同未设置，不生成 ENV_VARS 子句         |
| env_vars='{"K":"V"}'      | JSON 字符串直通                    |
| env_vars='not json'       | TypeError: must be valid JSON |
| env_vars=None             | 默认值，不生成 ENV_VARS 子句           |
   
#### Key格式约束
env_vars的key必须符合POSIX规范\^\[a-zA-Z_\]\[a-zA-Z0-9_\]\*$，最大长度1022字节。内核侧校验不合法时返回ERROR。
| Key 示例          | 合法 | 说明          |
|:---|:---|:---|
| MY_KEY          | 是  | 标准POSIX key |
| _UNDERSCORE_KEY | 是  | 以下划线开头      |
| KEY123          | 是  | 包含数字但非首字符   |
| 1KEY            | 否  | 以数字开头       |
| MY-KEY          | 否  | 包含连字符       |
| MY.KEY          | 否  | 包含点号        |
| MY=KEY          | 否  | 包含等号        |
| my key          | 否  | 包含空格        |
   
#### 安全注意事项
**禁止在env_vars中传入AK/SK等敏感凭证**
```
# 禁止 — 凭证明文经过 SQL 传输，有泄漏风险
my_udf(t.col).with_arguments(
    env_vars={"ACCESS_KEY_ID": "xxx", "SECRET_ACCESS_KEY": "yyy"}
)
# 禁止 — UDF_TEMP_AK/UDF_TEMP_SK是服务端凭证保留 key，用户设置会触发ERROR
my_udf(t.col).with_arguments(
    env_vars={"UDF_TEMP_AK": "xxx", "UDF_TEMP_SK": "yyy"}
)
# ERROR: env_vars key "UDF_TEMP_AK" conflicts with reserved server credentials, UDF 调用终止
```
**使用内核IAM委托Token自动注入**
内核通过IAM委托Token自动获取临时AK/SK，注入为UDF_TEMP_AK、UDF_TEMP_SK、UDF_TEMP_SECURITY_TOKEN环境变量。OBS Endpoint等非凭证配置通过用户env_vars传入。UDF内部直接通过os.environ读取，配合esdk-obs-python SDK使用：
```
class ObsDataReader:
    """Read data from OBS.
    Credentials (UDF_TEMP_AK/SK/TOKEN) are automatically injected by
    the kernel via IAM delegation token — no client-side setup needed.
    UDF_OBS_ENDPOINT must be provided by the user via env_vars.
    """
    def __init__(self, bucket: str):
        self.bucket = bucket
    def _get_obs_config(self) -> tuple:
        """Read OBS config from env vars."""
        import os
        # Credentials — auto-injected by kernel, do NOT set these yourself
        ak = os.environ.get("UDF_TEMP_AK", "")
        sk = os.environ.get("UDF_TEMP_SK", "")
        token = os.environ.get("UDF_TEMP_SECURITY_TOKEN", "")
        # Endpoint — user-provided via env_vars
        endpoint = os.environ.get("UDF_OBS_ENDPOINT", "")
        if not ak or not sk:
            raise RuntimeError(
                "OBS credentials not available. "
                "Ensure IAM delegation token is configured on the server."
            )
        if not endpoint:
            raise RuntimeError(
                "OBS endpoint not configured. "
                "Pass it via env_vars: with_arguments(env_vars={'UDF_OBS_ENDPOINT': '...'})"
            )
        return ak, sk, token, endpoint
    def __call__(self, object_key: str) -> str:
        """Read an object from OBS and return its content as string."""
        from obs import ObsClient
        ak, sk, token, endpoint = self._get_obs_config()
        obs_client = ObsClient(
            access_key_id=ak,
            secret_access_key=sk,
            security_token=token,
            server=endpoint,
        )
        try:
            resp = obs_client.getObject(
                self.bucket, object_key, loadStreamInMemory=True
            )
            if resp.status >= 300:
                raise RuntimeError(
                    f"OBS read failed: status={resp.status}, reason={resp.reason}"
                )
            return resp.body.buffer.decode("utf-8")
        finally:
            obs_client.close()
```
**使用方式** ：
```
import pandas as pd
from aura_frame.multimodal import ai_lake
con = ai_lake.connect(
    aura_endpoint="*",
    aura_workspace_id="*",
    lf_catalog_name="*",
    access_key="*",
    secret_key="*",
    default_database="my_schema",
    use_single_cn_mode=True,
)
# Register the UDFs
try:
    con.create_scalar_function(ObsDataReader, name="obs_reader", database="my_schema")
    # --- Direct expression API ---
    t = con.from_pandas(pd.DataFrame({"object_key": [""]}))._table
    result = con.get_function("obs_reader", database="my_schema")(
        t.object_key
    ).with_arguments(
        bucket="my-data-bucket",
        env_vars={"UDF_OBS_ENDPOINT": "obs.cn-north-4.myhuaweicloud.com"},
    )
    # --- Dataset API ---
    ds = con.from_pandas(pd.DataFrame({"object_key": [""]}))
    ds = ds.map(
        fn="my_schema.obs_reader",
        on=["object_key"],
        bucket="my-data-bucket",
        as_col="content",
        env_vars={"UDF_OBS_ENDPOINT": "obs.cn-north-4.myhuaweicloud.com"},
    )
    ds.show()
finally:
    con.close()
```
**服务端凭证保留key**（用户env_vars中设置会触发ERROR）：
| 保留 key                  | 用途                                           |
|:---|:---|
| UDF_TEMP_AK             | 临时Access Key。                                |
| UDF_TEMP_SK             | 临时Secret Key。                                |
| UDF_TEMP_SECURITY_TOKEN | 临时Security Token。                            |
| SHARED_WORK_DIR         | 会话级共享存储的路径，同一个session内的所有UDF实例都可以访问该路径，进行读写。 |
   
这些保留key由内核自动注入，用户不能手动设置。
**风险说明**：
- 前端env_vars以明文JSON通过SQL传输。
- 可能被审计日志、慢查询日志、错误日志捕获。
- 预置环境变量与用户key冲突时，内核返回ERROR并终止UDF调用。
**覆盖系统变量风险**：覆盖PATH、LD_PRELOAD、PYTHONPATH、HOME、PYTHONHOME 等可能导致非预期行为。