# coding: utf-8
#
# Copyright 2026 祁筱欣
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import os
import warnings
import threading
from abc import ABC, abstractmethod
from typing import Protocol, Type, Any
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkcore.client import Client
# 定义客户端构建器接口
class ClientBuilder(Protocol):
"""华为云客户端构建器接口"""
def with_credentials(self, credentials: BasicCredentials) -> "ClientBuilder":
"""设置认证信息"""
...
def with_region(self, region: Any) -> "ClientBuilder":
"""设置区域"""
...
def build(self) -> Client:
"""构建客户端实例"""
...
# 定义客户端类接口
class ClientClass(Protocol):
"""华为云客户端类接口"""
@staticmethod
def new_builder() -> ClientBuilder:
"""创建客户端构建器"""
...
# 定义区域类接口
class RegionClass(Protocol):
"""华为云区域类接口"""
@staticmethod
def value_of(region_id: str) -> Any:
"""根据区域ID获取区域对象"""
...
# 全局客户端实例缓存
_client_cache: dict[str, Client] = {}
# 全局锁对象，用于线程安全的客户端创建
_client_lock = threading.Lock()
class HuaweiCloudClient(ABC):
"""华为云客户端基类
提供通用的客户端创建和管理逻辑。
子类需要实现 get_client_class 和 get_region_class 方法。
"""
@abstractmethod
def get_client_class(self) -> Type[ClientClass]:
"""获取客户端类
Returns:
华为云客户端类（如 OcrClient、ModerationClient）
"""
...
@abstractmethod
def get_region_class(self) -> Type[RegionClass]:
"""获取区域类
Returns:
华为云区域类（如 OcrRegion、ModerationRegion）
"""
...
@abstractmethod
def get_cache_key(self) -> str:
"""获取缓存键
Returns:
客户端缓存键（如 "ocr"、"moderation"）
"""
...
def get_client(self) -> Client:
"""获取华为云客户端（线程安全的单例模式）
使用双重检查锁定（Double-Checked Locking）模式确保线程安全。
优先从环境变量获取，如果不存在则使用默认值。
Returns:
华为云客户端实例
Raises:
ValueError: 当环境变量未设置时抛出
"""
global _client_cache, _client_lock
cache_key = self.get_cache_key()
# 第一次检查：无锁，快速路径
if cache_key in _client_cache:
return _client_cache[cache_key]
# 获取锁，进行双重检查锁定
with _client_lock:
# 第二次检查：持有锁时再次检查缓存
# 防止多个线程同时通过第一次检查后重复创建客户端
if cache_key not in _client_cache:
# 从环境变量获取认证信息
ak = os.getenv("HUAWEI_CLOUD_SECRET_ID", "")
sk = os.getenv("HUAWEI_CLOUD_SECRET_KEY", "")
region = os.getenv("HUAWEI_CLOUD_REGION", "cn-east-3")
# 检查是否使用了默认值
if not ak or not sk:
if not ak:
warnings.warn("HUAWEI_CLOUD_SECRET_ID 未设置，请配置环境变量")
if not sk:
warnings.warn("HUAWEI_CLOUD_SECRET_KEY 未设置，请配置环境变量")
raise ValueError(
"环境变量 HUAWEI_CLOUD_SECRET_ID 和 HUAWEI_CLOUD_SECRET_KEY 未设置"
)
# 创建客户端
credentials = BasicCredentials(ak, sk)
client = (
self.get_client_class().new_builder()
.with_credentials(credentials)
.with_region(self.get_region_class().value_of(region))
.build()
)
# 缓存客户端
_client_cache[cache_key] = client
return _client_cache[cache_key]
