fiuai-sdk-python 0.11.0

FiuAI Python SDK - 企业级AI服务集成开发工具包

fiuai-sdk-python

FiuAI Python SDK — 企业级 AI 服务集成开发工具包。

提供统一的 FiuAI 平台接入能力, 包括: 认证上下文管理、Frappe API 客户端、HTTP 客户端工厂、Redis 缓存 (含熔断降级)、雪花 ID 生成、日志初始化等。

安装

# uv (推荐)
uv add fiuai-sdk-python

# pip
pip install fiuai-sdk-python

要求 Python >= 3.12

---

目录

• 快速开始
• SDK 初始化
• Frappe API 客户端
• 认证与上下文
• HTTP 客户端
• Redis 连接池管理
• 缓存客户端
• 缓存装饰器
• 熔断器
• 日志系统
• 雪花 ID 生成
• 工具函数
• FastAPI 集成
• API 参考索引

---

快速开始

from fiuai_sdk_python import init_fiuai, get_client

# 1. 初始化 SDK (全局一次)
init_fiuai(url="https://your-frappe-instance.com")

# 2. 获取客户端
client = get_client(
    username="[e-mail]",
    auth_tenant_id="tenant-001",
    current_company="company-001",
)

# 3. 调用 API
response = client.get_user_profile_info()
if response.is_success():
    print(response.data)

---

SDK 初始化

from fiuai_sdk_python import init_fiuai

init_fiuai(
    url="https://your-frappe-instance.com",
    max_api_retry=3,   # API 重试次数
    timeout=5,          # 请求超时 (秒)
    verify=False,       # SSL 证书校验
)

init_fiuai 是全局单例, 只需在应用启动时调用一次。后续通过 get_client() 获取客户端时, 会自动读取此配置。

---

Frappe API 客户端

获取客户端

from fiuai_sdk_python import get_client, FiuaiSDK

# 方式 1: 工厂函数 (推荐, 自动读取 init_fiuai 配置)
client = get_client(
    username="[e-mail]",
    auth_tenant_id="tenant-001",
    current_company="company-001",
    company_unique_no="UNQ-001",
)

# 方式 2: 直接实例化
client = FiuaiSDK(
    url="https://your-frappe-instance.com",
    username="[e-mail]",
    auth_tenant_id="tenant-001",
    current_company="company-001",
)

CRUD 操作

# 创建文档
resp = client.internal_create(data={
    "doctype": "Sales Invoice",
    "customer": "CUST-001",
    "items": [{"item_code": "ITEM-001", "qty": 10}],
})

# 查询单条
resp = client.internal_get("Sales Invoice", "SI-00001")

# 查询列表
resp = client.internal_get_list(
    doctype="Sales Invoice",
    filters={"customer": "CUST-001"},
    fields=["name", "grand_total"],
    limit_page_length=50,
    order_by="creation desc",
)

# 更新
resp = client.internal_update(data={
    "doctype": "Sales Invoice",
    "name": "SI-00001",
    "custom_field": "new_value",
})

# 删除
resp = client.internal_delete("Sales Invoice", "SI-00001")

# 提交 / 取消
resp = client.internal_submit("Sales Invoice", "SI-00001")
resp = client.internal_cancel("Sales Invoice", "SI-00001")

通用请求

# POST
resp = client.internal_post_req("/api/method/custom_method", postdata={"key": "value"})

# GET
resp = client.internal_get_req("/api/method/custom_method", params={"key": "value"})

临时头覆盖

跨租户/跨公司调用时, 可临时切换身份:

client.set_temp_header(
    auth_tenant_id="other-tenant",
    auth_company_id="other-company",
    user_id="[e-mail]",
)

resp = client.internal_get_list("Customer", limit_page_length=10)
client.clear_temp_headers()

响应处理

所有 API 调用返回 ApiResponse:

resp = client.internal_get("Sales Invoice", "SI-00001")

resp.http_success     # HTTP 状态码 2xx
resp.api_success      # 业务层成功
resp.is_success()     # http_success and api_success
resp.status_code      # HTTP 状态码
resp.data             # 响应数据
resp.error_code       # 错误码
resp.error_message    # 错误信息

---

认证与上下文

数据模型

from fiuai_sdk_python.auth import AuthData, AuthHeader

# AuthData: 解析后的认证数据
# 字段: user_id, auth_tenant_id, current_company, impersonation,
#       company_unique_no, trace_id, client, channel, lang, accept_language

# AuthHeader: HTTP 头映射
# 字段: x_fiuai_user, x_fiuai_auth_tenant_id, x_fiuai_current_company, ...

从请求解析认证

from fiuai_sdk_python.auth import parse_auth_headers, extract_auth_from_request

# 从字典解析
auth = parse_auth_headers({"X-Fiuai-User": "[e-mail]", ...})

# 从 FastAPI Request 解析
auth = extract_auth_from_request(request, engine="fastapi")

便捷函数

from fiuai_sdk_python.auth import (
    get_auth_data,
    get_current_user_id,
    get_current_tenant_id,
    get_current_company,
    get_company_unique_no,
    is_impersonating,
)

user_id = get_current_user_id(request)
tenant_id = get_current_tenant_id(request)
company_id = get_current_company(request)

上下文管理

from fiuai_sdk_python.auth import init_context, get_auth_data_from_context, ContextManager

# 初始化上下文 (通常在中间件中)
ctx = init_context(
    auth_data=auth_data,
    event_id="evt-001",
    task_id="task-001",
)

# 在业务代码中读取
auth = get_auth_data_from_context()

RequestContext (基于 contextvars)

from fiuai_sdk_python import RequestContext, get_current_headers, get_trace_id

# 注入上下文
with RequestContext.from_fastapi_request(request):
    headers = get_current_headers()
    trace_id = get_trace_id()

---

HTTP 客户端

提供基于 httpx 的 HTTP 客户端工厂, 支持自动注入认证头和重试。

from fiuai_sdk_python.http import (
    create_http_client,       # 创建异步客户端
    create_sync_http_client,  # 创建同步客户端
    get_async_http_client,    # 全局单例异步客户端
    get_sync_http_client,     # 全局单例同步客户端
    close_global_clients,     # 关闭全局客户端
)

# 异步客户端 (自动注入 auth headers)
async_client = create_http_client(
    base_url="https://api.example.com",
    timeout=30,
    retry_count=3,
    retry_interval=1,
)

async with async_client as client:
    resp = await client.get("/api/resource")

# 同步客户端
sync_client = create_sync_http_client(base_url="https://api.example.com")
resp = sync_client.get("/api/resource")

认证头拦截器

HTTP 客户端会自动从当前 RequestContext 中提取认证头并注入请求:

from fiuai_sdk_python.http import extract_auth_headers, AUTH_HEADER_KEYS

headers = extract_auth_headers()  # 从当前上下文提取

---

Redis 连接池管理

redis_manager 是全局单例, 管理多个 Redis 连接池。

初始化

from fiuai_sdk_python.pkg.cache import redis_manager, RedisDBConfig

await redis_manager.initialize(
    async_dbs=[
        RedisDBConfig(
            name="default",
            host="127.0.0.1",
            port=6379,
            password="your-password",
            db=0,
            pool_size=20,
        ),
        RedisDBConfig(
            name="cache",
            host="127.0.0.1",
            port=6379,
            password="your-password",
            db=1,
            pool_size=10,
        ),
    ],
    sync_dbs=[
        RedisDBConfig(
            name="default",
            host="127.0.0.1",
            port=6379,
            password="your-password",
            db=0,
            pool_size=5,
        ),
    ],
)

支持增量注册: 多次调用 initialize() 不会重复创建已存在的连接池。 async 和 sync 连接池独立追踪, 同名 db 可分别注册为 async 和 sync。

获取客户端

# 异步
async_client = redis_manager.get_async_client("default")
value = await async_client.get("key")

# 同步
sync_client = redis_manager.get_sync_client("default")
value = sync_client.get("key")

LangGraph Checkpoint (Agent 场景)

# 仅 Agent 项目需要, lazy import langgraph-checkpoint-redis
redis_manager.setup_langgraph_checkpoint("default")

# 异步版
await redis_manager.setup_langgraph_checkpoint_async("default")

关闭

await redis_manager.close_all()

---

缓存客户端

CacheClient 在 Redis 客户端之上提供: key 前缀管理、TTL、Hash 操作、cache-aside 模式、熔断降级。

创建

from fiuai_sdk_python.pkg.cache import CacheClient, CacheConfig, CircuitBreakerConfig

client = CacheClient(CacheConfig(
    redis_db_name="cache",          # redis_manager 中注册的连接池名
    default_ttl=300,                # 默认 TTL (秒), None 不过期
    key_prefix="myapp",             # key 前缀, 实际 key 为 "myapp:{key}"
    circuit_breaker=CircuitBreakerConfig(
        failure_threshold=5,        # 连续失败 5 次后熔断
        recovery_timeout=30.0,      # 熔断 30 秒后进入半开
        half_open_max_calls=1,      # 半开状态允许 1 次探测
    ),
))

KV 操作

# 异步
await client.set("user:123", '{"name": "test"}', ttl=60)
value = await client.get("user:123")        # -> Optional[str]
exists = await client.exists("user:123")     # -> bool
await client.delete("user:123")

# 同步
client.set_sync("user:123", '{"name": "test"}', ttl=60)
value = client.get_sync("user:123")

Hash 操作

await client.hset("config", "theme", "dark")
theme = await client.hget("config", "theme")    # -> "dark"
all_config = await client.hgetall("config")      # -> {"theme": "dark", ...}

Cache-Aside (核心)

get_or_load 是最常用的模式: 命中缓存直接返回, miss 时调用 loader 加载并写回缓存。

# 异步
async def load_user_config(user_id: str) -> dict:
    return await db.query("SELECT * FROM config WHERE user_id = ?", user_id)

config = await client.get_or_load(
    key=f"config:{user_id}",
    loader=lambda: load_user_config(user_id),
    ttl=120,
    # serializer/deserializer 默认 json.dumps/json.loads
)

# 同步
config = client.get_or_load_sync(
    key=f"config:{user_id}",
    loader=lambda: db.query_sync(...),
    ttl=120,
)

熔断降级

Redis 不可用时, CacheClient 自动降级:

• get / get_sync → 返回 None
• set / set_sync → 返回 False
• get_or_load / get_or_load_sync → 直接调 loader, 业务不中断

无需业务代码感知 Redis 故障。

---

缓存装饰器

@cached 装饰器将 cache-aside 模式封装为函数级注解。

初始化默认客户端

from fiuai_sdk_python.pkg.cache import init_cache, CacheConfig

# 应用启动时调用一次 (在 redis_manager.initialize 之后)
init_cache(CacheConfig(
    redis_db_name="cache",
    default_ttl=120,
    key_prefix="myapp",
))

使用装饰器

from fiuai_sdk_python.pkg.cache import cached

# 固定 key
@cached(key="global_settings", ttl=300)
async def get_settings() -> dict:
    return await fetch_settings_from_db()

# 动态 key (根据函数参数生成)
@cached(
    key=lambda company_id, side: f"config:{company_id}:{side}",
    ttl=120,
)
def get_company_config(company_id: str, side: str) -> dict:
    return query_config(company_id, side)

# 带 prefix
@cached(
    key=lambda user_id: f"profile:{user_id}",
    ttl=60,
    prefix="finnexus",       # 最终 key: "finnexus:profile:{user_id}"
)
async def get_user_profile(user_id: str) -> dict:
    ...

# 指定自定义 CacheClient
@cached(key="custom", client=my_cache_client)
def get_data() -> dict:
    ...

装饰器自动识别 sync/async 函数, 无需区分。

如果 init_cache 未调用 (即默认客户端为 None), 装饰器退化为直接调用原函数, 不会报错。

---

熔断器

CircuitBreaker 实现三态熔断 (CLOSED → OPEN → HALF_OPEN → CLOSED), 通常由 CacheClient 内部使用, 也可独立使用。

正常 (CLOSED) ──连续失败达阈值──→ 熔断 (OPEN)
       ↑                              │
       │                         冷却期过后
       │                              ↓
       └───── 探测成功 ───── 半开 (HALF_OPEN)
                                      │
                               探测失败 ──→ 回到 OPEN

独立使用

from fiuai_sdk_python.pkg.cache import CircuitBreaker, CircuitBreakerConfig

breaker = CircuitBreaker(CircuitBreakerConfig(
    failure_threshold=3,
    recovery_timeout=10.0,
    half_open_max_calls=1,
))

if breaker.allow_request:
    try:
        result = call_external_service()
        breaker.record_success()
    except Exception:
        breaker.record_failure()
else:
    result = fallback_value()

# 查看状态
print(breaker.state)  # CircuitState.CLOSED / OPEN / HALF_OPEN

---

日志系统

from fiuai_sdk_python.utils.logger import init_logger, get_logger

# 初始化 (应用启动时调用一次)
init_logger(
    log_path="logs/",
    log_level="INFO",
    context_injector=lambda: {"trace_id": get_current_trace_id()},
)

# 在各模块中使用
logger = get_logger(__name__)
logger.info("processing request")
logger.error("something went wrong", exc_info=True)

日志格式: %(asctime)s - %(name)s - %(levelname)s - [trace_id:%(trace_id)s] - %(message)s

context_injector 可注入自定义上下文字段 (如 trace_id) 到每条日志。

---

雪花 ID 生成

from fiuai_sdk_python.utils.ids import gen_id, get_instance_info

# 生成唯一 ID
unique_id = gen_id()  # -> "7345892348923489234" (字符串)

# 查看实例信息
info = get_instance_info()
# -> {"instance_id": 42, "process_id": 12345, "start_time": ...}

# 自定义实例 ID (分布式部署时)
from fiuai_sdk_python.utils.ids import set_custom_instance_id
set_custom_instance_id(42)

---

工具函数

文本处理

from fiuai_sdk_python.utils.text import safe_string_name, safe_str, safe_json_str

# 全角符号转半角
safe_string_name("北京（朝阳）公司")  # -> "北京(朝阳)公司"

# URL 安全编码 (用于 Redis 密码等含特殊字符的场景)
safe_str("p@ss#word!")  # -> "p%40ss%23word%21"

# JSON 安全处理 (转义控制字符)
safe_json_str("line1\nline2")  # -> "line1\\nline2"

---

FastAPI 集成

中间件注入上下文

from fastapi import FastAPI
from fiuai_sdk_python import init_fiuai
from fiuai_sdk_python.examples.fastapi_integration import setup_fiuai_context

app = FastAPI()

# 初始化 SDK
init_fiuai(url="https://your-frappe-instance.com")

# 注册上下文中间件 (自动将请求头注入 RequestContext)
setup_fiuai_context(app)

完整启动示例

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fiuai_sdk_python import init_fiuai
from fiuai_sdk_python.utils.logger import init_logger
from fiuai_sdk_python.pkg.cache import redis_manager, RedisDBConfig, init_cache, CacheConfig
from fiuai_sdk_python.examples.fastapi_integration import setup_fiuai_context


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动
    init_logger(log_path="logs/", log_level="INFO")
    init_fiuai(url="https://your-frappe-instance.com")

    await redis_manager.initialize(async_dbs=[
        RedisDBConfig(name="default", host="127.0.0.1", port=6379, password="", db=0),
    ])
    init_cache(CacheConfig(redis_db_name="default", default_ttl=120, key_prefix="myapp"))

    yield

    # 关闭
    await redis_manager.close_all()


app = FastAPI(lifespan=lifespan)
setup_fiuai_context(app)

---

API 参考索引

顶层导出 (fiuai_sdk_python)

名称	类型	说明
init_fiuai	function	初始化 SDK 全局配置
FiuaiSDK	class	Frappe API 客户端
get_client	function	客户端工厂
UserProfileInfo	class	用户信息模型
UserProfile	class	用户基本模型
RequestContext	class	请求上下文 (contextvars)
get_current_headers	function	获取当前上下文 headers
get_trace_id	function	获取当前 trace_id

认证模块 (fiuai_sdk_python.auth)

名称	类型	说明
AuthData	class	解析后的认证数据
AuthHeader	class	HTTP 认证头映射
parse_auth_headers	function	从 dict 解析认证
extract_auth_from_request	function	从 Request 解析认证
init_context	function	初始化上下文管理器
get_auth_data_from_context	function	从上下文获取 AuthData
ContextManager	class	上下文管理器
WorldData	class	World 事件/任务上下文

HTTP 模块 (fiuai_sdk_python.http)

名称	类型	说明
create_http_client	function	创建异步 HTTP 客户端
create_sync_http_client	function	创建同步 HTTP 客户端
get_async_http_client	function	全局异步客户端单例
get_sync_http_client	function	全局同步客户端单例
close_global_clients	function	关闭全局客户端
extract_auth_headers	function	从上下文提取认证头

缓存模块 (fiuai_sdk_python.pkg.cache)

名称	类型	说明
redis_manager	RedisManager	Redis 连接池管理器 (单例)
RedisDBConfig	class	Redis 连接配置
CacheClient	class	通用缓存客户端
CacheConfig	class	缓存客户端配置
init_cache	function	初始化默认 CacheClient
get_default_client	function	获取默认 CacheClient
cached	decorator	函数级缓存装饰器
CircuitBreaker	class	三态熔断器
CircuitBreakerConfig	class	熔断器配置
CircuitState	enum	熔断状态枚举

工具模块 (fiuai_sdk_python.utils)

名称	模块	说明
init_logger	logger	初始化日志系统
get_logger	logger	获取 logger 实例
safe_string_name	text	全角转半角
safe_str	text	URL 安全编码
safe_json_str	text	JSON 安全处理
gen_id	ids	生成雪花 ID
set_custom_instance_id	ids	设置自定义实例 ID

---

版本

当前版本: 0.8.1

License

MIT License - Copyright (c) 2025 FiuAI