huace-aigc-auth-client 1.1.45

华策AIGC Auth Client - 提供 Token 验证、用户信息获取、权限检查、旧系统接入等功能

AIGC Auth Python SDK

Python 后端服务接入华策 AIGC 鉴权中心的 SDK 工具包。

安装

pip install huace-aigc-auth-client

环境变量配置

在项目根目录创建 .env 文件：

# 必填：应用 ID 和密钥（在鉴权中心创建应用后获取）
AIGC_AUTH_APP_ID=your_app_id
AIGC_AUTH_APP_SECRET=your_app_secret
# 鉴权服务地址（默认为生产环境）
AIGC_AUTH_BASE_URL=your-auth-api-url-prefix

如需通过 Nginx 代理：

location /aigc-auth/ {
    proxy_pass https://aigc-auth.huacemedia.com/aigc-auth/;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
}

快速开始

1. 初始化客户端

from huace_aigc_auth_client import AigcAuthClient

# 方式一：从环境变量读取配置
client = AigcAuthClient()

# 方式二：直接传入参数
client = AigcAuthClient(
    app_id="your_app_id",
    app_secret="your_app_secret",
    base_url="your-auth-api-url-prefix"
)

2. 验证 Token

result = client.verify_token(token)

if result.valid:
    print(f"用户 ID: {result.user_id}")
    print(f"用户名: {result.username}")
    print(f"过期时间: {result.expires_at}")
else:
    print("Token 无效")

3. 获取用户信息

from huace_aigc_auth_client import AigcAuthClient, AigcAuthError

client = AigcAuthClient()

try:
    user = client.get_user_info(token)
    
    print(f"用户名: {user.username}")
    print(f"昵称: {user.nickname}")
    print(f"邮箱: {user.email}")
    print(f"角色: {user.roles}")
    print(f"权限: {user.permissions}")
    
    # 检查角色(在 Auth 里面配置)
    if user.has_role("admin") or user.is_admin:
        print("是管理员")
    
    # 检查权限(在 Auth 里面配置)
    if user.has_permission("user:write"):
        print("有用户写权限")
        
except AigcAuthError as e:
    print(f"错误: {e.message}")

4. 批量检查权限

results = client.check_permissions(token, ["user:read", "user:write", "admin:access"])

for permission, has_permission in results.items():
    print(f"{permission}: {'✓' if has_permission else '✗'}")

5. 批量获取用户信息

# 按用户名批量查询
result = client.batch_get_users(["zhangsan", "lisi"])
for user in result["users"]:
    print(f"{user['username']}: {user['nickname']}")
# result["not_found"] 包含未找到的用户名

# 按用户ID批量查询（未找到的返回"未知用户"占位）
result = client.batch_get_users_by_ids([1, 2, 999])
for user in result["users"]:
    print(f"{user['id']}: {user['username']}")

FastAPI 集成

方式一：使用中间件（推荐）

from fastapi import FastAPI, Request, HTTPException
from huace_aigc_auth_client import AigcAuthClient, AuthMiddleware

app = FastAPI()

# 初始化客户端和中间件
client = AigcAuthClient()
auth_middleware = AuthMiddleware(
    client,
    exclude_paths=["/health", "/docs", "/openapi.json"],
    exclude_prefixes=["/public/"]
)

# 注册中间件
@app.middleware("http")
async def auth(request: Request, call_next):
    return await auth_middleware.fastapi_middleware(request, call_next)

# 在路由中获取用户信息
@app.get("/me")
async def get_current_user(request: Request):
    user = request.state.user_info
    return {
        "username": user.username,
        "roles": user.roles
    }

@app.get("/admin")
async def admin_only(request: Request):
    user = request.state.user_info
    if not user.has_role("admin"):
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return {"message": "欢迎管理员"}

方式二：使用依赖注入

from fastapi import FastAPI, Depends, HTTPException
from huace_aigc_auth_client import AigcAuthClient, create_fastapi_auth_dependency, UserInfo

app = FastAPI()
client = AigcAuthClient()

# 创建认证依赖
get_current_user = create_fastapi_auth_dependency(client)

@app.get("/me")
async def get_me(user: UserInfo = Depends(get_current_user)):
    return {"username": user.username}

# 创建权限检查依赖
def require_permission(permission: str):
    async def check(user: UserInfo = Depends(get_current_user)):
        if not user.has_permission(permission):
            raise HTTPException(status_code=403, detail="权限不足")
        return user
    return check

@app.get("/users")
async def list_users(user: UserInfo = Depends(require_permission("user:read"))):
    return {"users": [...]}

方式三：使用装饰器

from fastapi import FastAPI, Request
from huace_aigc_auth_client import AigcAuthClient, require_auth, UserInfo

app = FastAPI()
client = AigcAuthClient()

@app.get("/protected")
@require_auth(client)
async def protected_route(request: Request, user_info: UserInfo):
    return {"user": user_info.username}

@app.get("/admin")
@require_auth(client, permissions=["admin:access"])
async def admin_route(request: Request, user_info: UserInfo):
    return {"admin": True}

# 只需要任意一个权限
@app.get("/editor")
@require_auth(client, permissions=["article:write", "article:edit"], any_permission=True)
async def editor_route(request: Request, user_info: UserInfo):
    return {"editor": True}

Flask 集成

from flask import Flask, g, jsonify
from functools import wraps
from huace_aigc_auth_client import AigcAuthClient, AuthMiddleware

app = Flask(__name__)

# 初始化客户端和中间件
client = AigcAuthClient()
auth_middleware = AuthMiddleware(
    client,
    exclude_paths=["/health", "/login"],
    exclude_prefixes=["/public/"]
)

# 注册 before_request
@app.before_request
def before_request():
    return auth_middleware.flask_before_request()

# 在路由中获取用户信息
@app.route("/me")
def get_me():
    user = g.user_info
    return jsonify({
        "username": user.username,
        "roles": user.roles
    })

# 权限检查装饰器
def require_permission(permission):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            user = g.user_info
            if not user.has_permission(permission):
                return jsonify({"error": "权限不足"}), 403
            return f(*args, **kwargs)
        return decorated_function
    return decorator

@app.route("/admin")
@require_permission("admin:access")
def admin_only():
    return jsonify({"message": "欢迎管理员"})

上下文管理

SDK 提供了上下文管理功能，可以在任意位置（Service、Utility 等）获取当前请求的用户信息，无需层层传递参数。 支持 Flask 和 FastAPI，兼容同步和异步环境。

注意：使用此功能前必须先注册 AuthMiddleware。

from huace_aigc_auth_client import get_current_user

def do_something_logic():
    # 获取当前用户信息（返回字典，非 UserInfo 对象）
    user = get_current_user()
    
    if user:
        print(f"当前操作用户ID: {user['id']}")
        print(f"当前操作用户名: {user['username']}")
        
        # 也可以获取 app_id 等信息（如果未过滤）
        if 'app_id' in user:
            print(f"应用ID: {user['app_id']}")
    else:
        print("无用户上下文")

API 参考

UserInfo 对象

属性	类型	说明
id	int	用户 ID
username	str	用户名
nickname	str	昵称
email	str	邮箱
phone	str	手机号
avatar	str	头像 URL
roles	List[str]	角色代码列表
permissions	List[str]	权限代码列表
department	str	部门
company	str	公司

方法	说明
has_role(role)	检查是否拥有指定角色
has_permission(permission)	检查是否拥有指定权限
has_any_permission(permissions)	检查是否拥有任意一个权限
has_all_permissions(permissions)	检查是否拥有所有权限

异常处理

from huace_aigc_auth_client import AigcAuthClient, AigcAuthError

client = AigcAuthClient()

try:
    user = client.get_user_info(token)
except AigcAuthError as e:
    print(f"错误码: {e.code}")
    print(f"错误信息: {e.message}")

错误码	说明
401	未授权（Token 无效或已过期）
403	禁止访问（用户被禁用或权限不足）
404	用户不存在
-1	网络请求失败

---

旧系统接入（用户同步）

如果你的系统已有用户表，可通过 SDK 提供的「旧系统适配器」实现低成本接入，无需修改历史代码和表结构。

接入原理

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   aigc-auth     │────▶│   SDK 同步层     │────▶│   旧系统         │
│   (鉴权中心)     │     │   (字段映射)     │     │   (用户表)       │
└─────────────────┘     └─────────────────┘     └─────────────────┘
        │                       │                       │
        │   1. 初始化同步        │                       │
        │◀──────────────────────────────────────────────│
        │   (批量同步旧用户到 auth)                       │
        │                       │                       │
        │   2. 增量同步          │                       │
        │──────────────────────▶│──────────────────────▶│
        │   (auth 新用户自动同步到旧系统)                  │
        │                       │                       │
        │   3. Webhook 推送      │                       │
        │──────────────────────────────────────────────▶│
        │   (用户变更主动通知)                            │

快速开始

1. 配置环境变量

# .env 文件
AIGC_AUTH_APP_ID=your_app_id
AIGC_AUTH_APP_SECRET=your_app_secret
AIGC_AUTH_BASE_URL=your-auth-api-url-prefix

# 同步配置
AIGC_AUTH_SYNC_ENABLED=true
AIGC_AUTH_SYNC_PASSWORD=通用密码
AIGC_AUTH_WEBHOOK_URL=https://your-domain.com/api/v1/webhook/auth
AIGC_AUTH_WEBHOOK_SECRET=your_secret

2. 创建字段映射

from huace_aigc_auth_client import (
    FieldMapping,
    SyncConfig,
    PasswordMode,
    create_sync_config
)

# 定义字段映射
field_mappings = [
    FieldMapping(
        auth_field="username",
        legacy_field="username",
        required=True
    ),
    FieldMapping(
        auth_field="email",
        legacy_field="email"
    ),
    FieldMapping(
        auth_field="nickname",
        legacy_field="nickname"
    ),
    # 状态字段转换：auth 的 status 映射到旧系统的 is_active
    FieldMapping(
        auth_field="status",
        legacy_field="is_active",
        transform_to_legacy=lambda s: s == "active",
        transform_to_auth=lambda a: "active" if a else "disabled"
    ),
    # 角色字段转换：auth 的 roles 列表映射到旧系统的 role 字符串
    FieldMapping(
        auth_field="roles",
        legacy_field="role",
        transform_to_legacy=lambda r: r[0] if r else "viewer",
        transform_to_auth=lambda r: [r] if r else ["viewer"]
    ),
]

# 创建同步配置
sync_config = create_sync_config(
    field_mappings=field_mappings,
    password_mode=PasswordMode.UNIFIED,
    unified_password="通用密码",
    webhook_url="https://your-domain.com/api/v1/webhook/auth",
)

3. 实现旧系统适配器

from huace_aigc_auth_client import LegacySystemAdapter, LegacyUserData

class MyLegacyAdapter(LegacySystemAdapter):
    """实现与旧系统用户表的交互"""
    
    def __init__(self, db, sync_config, auth_client=None):
        super().__init__(sync_config, auth_client)
        self.db = db
    
    def get_user_by_unique_field(self, username: str):
        """通过用户名获取旧系统用户"""
        user = self.db.query(User).filter(User.username == username).first()
        if not user:
            return None
        return LegacyUserData({
            "id": user.id,
            "username": user.username,
            "email": user.email,
            # ... 其他字段
        })

    async def get_user_by_username_async(self, username: str) -> Optional[LegacyUserData]:
        """异步通过用户名获取旧系统用户"""
        user = await self.db.execute(
            select(User).where(User.username == username)
        )
        user = user.scalars().first()
        if not user:
            return None
        return LegacyUserData({
            "id": user.id,
            "username": user.username,
            "email": user.email,
            # ... 其他字段
        })
    
    async def _create_user_async(self, user_data: Dict[str, Any]) -> Optional[Any]:
        """在旧系统创建用户"""
        user = User(
            username=user_data["username"],
            email=user_data.get("email"),
            hashed_password=hash_password(user_data["password"]),
            # ... 其他字段
        )
        self.db.add(user)
        await self.db.commit()
        return user.id
    
    async def _update_user_async(self, username: str, user_data: Dict[str, Any]) -> bool:
        """异步更新旧系统用户"""
        user = await self.db.execute(
            select(User).where(User.username == username)
        )
        user = user.scalars().first()
        if user:
            for key, value in user_data.items():
                setattr(user, key, value)
            await self.db.commit()
            return True
        return False

    async def _delete_user_async(self, username: str) -> bool:
        """异步删除旧系统用户"""
        user = await self.db.execute(
            select(User).where(User.username == username)
        )
        user = user.scalars().first()
        if user:
            await self.db.delete(user) 
            # 或者软删除：user.is_active = False
            await self.db.commit()
            return True
        return False
    
    async def get_all_users_async(self) -> List[LegacyUserData]:
        """异步获取所有用户（用于初始化同步）"""
        result = await self.db.execute(select(User))
        users = result.scalars().all()
        return [LegacyUserData({"username": u.username, ...}) for u in users]

4. 集成到登录流程

from huace_aigc_auth_client import AigcAuthClient, UserSyncService

client = AigcAuthClient()
adapter = MyLegacyAdapter(db, sync_config, client)
sync_service = UserSyncService(client, adapter)

# 在获取用户信息后调用同步
@app.middleware("http")
async def auth_middleware(request, call_next):
    response = await auth.fastapi_middleware(request, call_next)
    
    # 登录成功后，同步用户到旧系统
    if hasattr(request.state, "user_info"):
        user_info = request.state.user_info
        await sync_service.sync_on_login_async(user_info)
    
    return response

5. 添加 Webhook 接收端点

推荐方式：使用 SDK 提供的通用 Webhook 路由

from fastapi import APIRouter
from huace_aigc_auth_client import register_webhook_router

api_router = APIRouter()
client = AigcAuthClient()
adapter = MyLegacyAdapter(db, sync_config, client)

# 定义 webhook 处理函数
async def handle_auth_webhook(request_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    独立的 webhook 处理函数（用于 SDK 集成）
    
    这个函数不需要 db 参数，会在内部创建数据库会话。
    适用于通过 SDK 的 register_webhook_router 注册。
    """
    from app.db.session import get_db
    
    async for db in get_db():
        try:
            event = request_data.get("event")
            data = request_data.get("data", {})
            logger.info(f"Received webhook event: {event} with data: {data}")
            
            # 调用适配器的 handle_webhook 方法
            return await adapter.handle_webhook(event, data)
        except Exception as e:
            logger.exception(f"Failed to handle webhook: {e}")
            raise

# 注册 webhook 路由（自动处理签名验证）
register_webhook_router(
    api_router,
    handler=handle_auth_webhook,
    prefix="/webhook",  # 可选，默认 "/webhook"
    secret_env_key="aigc-auth-webhook-secret",  # 可选，在 auth 后台配置
    tags=["Webhook"]  # 可选，默认 ["Webhook"]
)

# webhook 端点将自动创建在: /webhook/auth

Webhook 签名验证说明

SDK 的 register_webhook_router 会自动处理签名验证：

• 从请求头 X-Webhook-Signature 获取签名
• 从环境变量读取密钥（默认 AIGC_AUTH_WEBHOOK_SECRET）
• 使用 HMAC-SHA256 算法验证签名
• 签名不匹配时返回 401 错误

6. 执行初始化同步

# 一次性脚本：将旧系统用户同步到 aigc-auth
async def init_sync():
    adapter = MyLegacyAdapter(db, sync_config, client)
    users = adapter.get_all_users()
    
    for user in users:
        auth_data = adapter.transform_legacy_to_auth(user)
        
        # 获取密码（支持元组返回格式）
        password_result = adapter.get_password_for_sync(user)
        if isinstance(password_result, tuple):
            password, is_hashed = password_result
        else:
            password, is_hashed = password_result, False
        
        # 根据是否已加密选择不同的字段
        if is_hashed:
            auth_data["password_hashed"] = password  # 直接传递已加密密码
        else:
            auth_data["password"] = password  # 传递明文密码，服务端会加密
        
        client.sync_user_to_auth(auth_data)
    
    print(f"同步完成：{len(users)} 个用户")

# 运行
asyncio.run(init_sync())

密码处理策略

模式	说明	适用场景
UNIFIED	统一初始密码	新系统接入，安全性高
CUSTOM_MAPPING	自定义映射函数	需要保留原密码时
CUSTOM_MAPPING + password_is_hashed=True	直接同步已加密密码	推荐，两系统使用相同加密方式 (bcrypt)

from huace_aigc_auth_client import PasswordMode, create_sync_config

# 统一密码模式
sync_config = create_sync_config(
    password_mode=PasswordMode.UNIFIED,
    unified_password="Abc@123456"
)

# 自定义映射模式（返回明文密码）
def password_mapper(user_data):
    return user_data.get("password", "default")

sync_config = create_sync_config(
    password_mode=PasswordMode.CUSTOM_MAPPING,
    password_mapper=password_mapper
)

# 直接同步已加密密码（推荐，适用于两系统使用相同的 bcrypt 加密）
def hashed_password_mapper(user_data):
    return user_data.get("hashed_password", "")

sync_config = create_sync_config(
    password_mode=PasswordMode.CUSTOM_MAPPING,
    password_mapper=hashed_password_mapper,
    password_is_hashed=True  # 标记返回的是已加密密码
)

注意：当 password_is_hashed=True 时，SDK 会使用 password_hashed 字段传递到服务端， 服务端会直接使用该哈希值而不再重新加密。

同步方向配置

from huace_aigc_auth_client import SyncDirection, create_sync_config

# 仅从 auth 同步到旧系统（默认）
sync_config = create_sync_config(
    direction=SyncDirection.AUTH_TO_LEGACY
)

# 仅从旧系统同步到 auth（初始化用）
sync_config = create_sync_config(
    direction=SyncDirection.LEGACY_TO_AUTH
)

# 双向同步（需谨慎使用）
sync_config = create_sync_config(
    direction=SyncDirection.BIDIRECTIONAL
)

---

高级功能

API 统计收集

SDK 提供了接口统计收集功能，自动收集和上报 API 调用数据到鉴权中心。

1. 初始化统计收集器

from huace_aigc_auth_client import init_api_stats_collector

# 仅仅示例，SDK 接入鉴权会默认启用数据收集功能（可以手动传参关闭）
# 在应用启动时初始化
init_api_stats_collector(
    api_url='http://auth.example.com/api/sdk',
    app_id='your-app-id',
    app_secret='your-app-secret',
    batch_size=10,          # 批量提交大小
    flush_interval=5.0,     # 刷新间隔（秒）
    enabled=True            # 是否启用
)

2. 在中间件中自动收集

FastAPI 示例：

from fastapi import FastAPI, Request
from huace_aigc_auth_client import collect_api_stat
import time

app = FastAPI()


# 仅仅示例，SDK 内默认会启用数据收集未 exclude 的接口（可以手动传参关闭）
@app.middleware("http") 
async def monitor_middleware(request: Request, call_next):
    start_time = time.time()
    
    try:
        response = await call_next(request)
        response_time = time.time() - start_time
        
        # 从请求上下文获取 token
        token = request.state.user_info.token if hasattr(request.state, 'user_info') else None
        
        # 收集统计（会自动过滤 3xx 重定向请求）
        collect_api_stat(
            api_path=request.url.path,
            api_method=request.method,
            status_code=response.status_code,
            response_time=response_time,
            token=token,
            request_params={
                'headers': dict(request.headers),
                'query_params': dict(request.query_params)
            }
        )
        
        return response
    except Exception as e:
        response_time = time.time() - start_time
        collect_api_stat(
            api_path=request.url.path,
            api_method=request.method,
            status_code=500,
            response_time=response_time,
            token=token,
            error_message=str(e)
        )
        raise

@app.on_event("shutdown")
async def shutdown_event():
    from huace_aigc_auth_client import stop_api_stats_collector
    stop_api_stats_collector()

特性说明：

• 异步队列收集，不阻塞主流程
• 批量提交，减少网络开销
• 自动过滤 3xx 重定向请求
• 按 token 分组提交
• 静默失败，不影响主业务

---

认证请求封装

SDK 提供了 auth_request 函数，封装标准的 HTTP 请求，自动添加认证信息并上报统计。

1. 基本使用

from huace_aigc_auth_client import auth_request, set_request_context

# 设置请求上下文（通常在中间件中完成）
set_request_context(
    app_id='your-app-id',
    app_secret='your-app-secret',
    token='user-access-token',
    ip_address='192.168.1.1',
    user_agent='Mozilla/5.0...',
    trace_id='trace-123'
)

# 发起请求（会自动添加认证信息）
response = auth_request('GET', 'https://api.example.com/data')
print(response.json())

2. 带参数的请求

# GET 请求带查询参数
response = auth_request(
    'GET',
    'https://api.example.com/users',
    params={'page': 1, 'size': 10}
)

# POST 请求带 JSON 数据
response = auth_request(
    'POST',
    'https://api.example.com/users',
    json={'name': 'John', 'email': '[e-mail]'}
)

# POST 请求带表单数据
response = auth_request(
    'POST',
    'https://api.example.com/upload',
    data={'field1': 'value1', 'field2': 'value2'}
)

3. 异步请求（httpx）

需要安装 httpx：pip install httpx

from huace_aigc_auth_client import async_auth_httpx_request, AsyncAuthClient
import asyncio

# 方式1：使用函数
async def example1():
    response = await async_auth_httpx_request('GET', 'https://api.example.com/data')
    print(response.json())

# 方式2：使用 AsyncAuthClient（推荐用于多个请求）
async def example2():
    async with AsyncAuthClient(timeout=10.0) as client:
        # POST 请求
        response = await client.post(
            'https://api.example.com/users',
            json={'name': 'John'}
        )
        
        # GET 请求
        response = await client.get(
            'https://api.example.com/users',
            params={'page': 1}
        )

asyncio.run(example1())

自动添加的请求头：

• X-App-ID: 应用ID
• X-App-Secret: 应用密钥
• Authorization: Bearer token
• x-real-ip: 客户端真实IP
• user-agent: User Agent
• X-Trace-ID: 追踪ID

---

用户上下文管理

SDK 提供了完整的用户上下文管理功能，支持同步和异步环境。

1. 设置和获取用户信息

from huace_aigc_auth_client import (
    set_current_user, 
    get_current_user,
    get_current_user_id,
    get_current_username,
    is_current_user_admin
)

# 在中间件中设置
set_current_user({
    'user_id': 123,
    'username': 'john',
    'app_id': 1,
    'app_code': 'myapp',
    'token': 'xxx',
    'roles': ['admin'],
    'permissions': ['user:read', 'user:write'],
    'is_admin': True
})

# 在业务代码中获取
def some_business_logic():
    user = get_current_user()
    user_id = get_current_user_id()
    username = get_current_username()
    
    if user:
        print(f"当前用户: {username} ({user_id})")
        
    if is_current_user_admin():
        print("当前用户是管理员")

2. 请求上下文管理

from huace_aigc_auth_client import (
    set_request_context,
    get_request_context,
    get_client_ip,
    get_user_agent,
    get_trace_id
)

# 设置请求上下文
set_request_context(
    ip_address='192.168.1.1',
    user_agent='Mozilla/5.0...',
    trace_id='trace-123',
    app_id='your-app-id',
    app_secret='your-app-secret',
    token='user-token'
)

# 获取请求上下文信息
ctx = get_request_context()
ip = get_client_ip()
ua = get_user_agent()
trace = get_trace_id()

3. 在 FastAPI 中集成

from fastapi import FastAPI, Request, Depends
from huace_aigc_auth_client import (
    set_current_user,
    set_request_context,
    get_current_user,
    clear_current_user,
    clear_request_context
)

app = FastAPI()

@app.middleware("http")
async def user_context_middleware(request: Request, call_next):
    # 从认证中获取用户信息
    user_info = request.state.user_info if hasattr(request.state, 'user_info') else None
    
    if user_info:
        # 设置用户上下文
        set_current_user({
            'user_id': user_info.id,
            'username': user_info.username,
            'app_id': getattr(user_info, 'app_id', None),
            'is_admin': user_info.is_admin
        })
        
        # 设置请求上下文
        set_request_context(
            ip_address=request.client.host,
            user_agent=request.headers.get('user-agent'),
            trace_id=request.headers.get('x-trace-id'),
            token=request.headers.get('authorization', '').replace('Bearer ', '')
        )
    
    try:
        response = await call_next(request)
        return response
    finally:
        clear_current_user()
        clear_request_context()

# 在任意地方使用
def my_service_function():
    user = get_current_user()
    if user:
        print(f"执行操作的用户: {user['username']}")

4. 装饰器中使用

from functools import wraps
from huace_aigc_auth_client import is_current_user_admin

def require_admin(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        if not is_current_user_admin():
            raise HTTPException(status_code=403, detail="需要管理员权限")
        return await func(*args, **kwargs)
    return wrapper

@require_admin
async def admin_endpoint():
    return {"message": "管理员操作"}

特性说明：

• 使用 threading.local() 和 contextvars.ContextVar 实现
• 同时支持同步和异步环境
• 线程安全和异步任务隔离
• 无需层层传递参数

---

导出清单

from huace_aigc_auth_client import (
    # 核心类
    AigcAuthClient,
    require_auth,
    AuthMiddleware,
    UserInfo,
    TokenVerifyResult,
    AigcAuthError,
    create_fastapi_auth_dependency,
    
    # 旧系统接入
    LegacySystemAdapter,
    LegacyUserData,
    SyncConfig,
    SyncDirection,
    PasswordMode,
    FieldMapping,
    UserSyncService,
    WebhookSender,
    SyncResult,
    create_sync_config,
    create_default_field_mappings,
    
    # Webhook 接收
    register_webhook_router,
    verify_webhook_signature,
    
    # API 统计收集
    ApiStatsCollector,
    init_api_stats_collector,
    get_api_stats_collector,
    stop_api_stats_collector,
    collect_api_stat,
    
    # 认证请求封装
    auth_request,
    async_auth_httpx_request,
    AsyncAuthClient,
    
    # 用户上下文管理
    set_current_user,
    get_current_user,
    get_current_user_id,
    get_current_username,
    get_current_app_id,
    get_current_app_code,
    is_current_user_admin,
    clear_current_user,
    set_request_context,
    get_request_context,
    get_client_ip,
    get_user_agent,
    get_request_id,
    get_trace_id,
    get_request_token,
    clear_request_context,
)

---5 (2026-01-15)

新增功能

1. Webhook 接收功能
   • 新增 register_webhook_router() 函数，快速注册 webhook 接收路由
   • 自动处理签名验证、请求解析和错误处理
   • 支持自定义前缀、密钥环境变量和标签
   • 新增 verify_webhook_signature() 工具函数

使用示例

from fastapi import APIRouter
from huace_aigc_auth_client import register_webhook_router

api_router = APIRouter()

async def my_handler(data: dict) -> dict:
    event = data.get("event")
    # 处理逻辑...
    return {"status": "ok"}

# 注册 webhook 路由
register_webhook_router(api_router, my_handler)

v1.1.

API 变更日志

v1.1.42 (2026-04-28)

新增功能

1. 批量获取用户信息（按用户名）

   • 新增 batch_get_users(usernames) 方法
   • 通过用户名列表批量查询用户，返回属于当前应用或全局用户（app_id 为 null）的用户信息
   • 未找到的用户名会汇总到 not_found 列表

2. 批量获取用户信息（按用户ID）

   • 新增 batch_get_users_by_ids(user_ids) 方法
   • 通过用户 ID 列表批量查询用户，按请求顺序返回结果
   • 未找到的用户返回 "未知用户" 占位信息

使用示例

# 按用户名批量查询
result = client.batch_get_users(["zhangsan", "lisi", "wangwu"])
for user in result["users"]:
    print(f"{user['username']}: {user['nickname']}")
print(f"未找到: {result['not_found']}")

# 按用户ID批量查询
result = client.batch_get_users_by_ids([1, 2, 3, 999])
for user in result["users"]:
    if user["username"] == "未知用户":
        print(f"用户 {user['id']} 不存在")
    else:
        print(f"{user['id']}: {user['username']}")

v1.1.0 (2026-01-13)

新增功能

1. 支持直接同步已加密密码

   • SyncConfig 新增 password_is_hashed 参数
   • SyncUserRequest 新增 password_hashed 字段
   • 当两个系统使用相同的密码加密方式（bcrypt）时，可直接同步已加密密码

2. _request 方法支持自定义 headers

   • sync_user_to_auth(user_data, headers=None) 支持传入自定义请求头
   • batch_sync_users_to_auth(users, headers=None) 支持传入自定义请求头

3. get_password_for_sync 返回格式变更

   • 旧版：返回 str（密码字符串）
   • 新版：返回 tuple[str, bool]（密码字符串, 是否已加密）

使用示例

# 直接同步已加密密码
sync_config = create_sync_config(
    password_mode=PasswordMode.CUSTOM_MAPPING,
    password_mapper=lambda user: user.get("hashed_password"),
    password_is_hashed=True,  # 新增参数
)

# SDK 调用
client.sync_user_to_auth({
    "username": "test",
    "password_hashed": "$2b$12$xxxxx...",  # 直接传递 bcrypt 哈希值
    "email": "[e-mail]"
})

许可证

MIT License