企业官网建设流程全解析

1. 先说清楚：affinidi-tdk-common 是什么，解决什么问题

如果你和我一样，最近在做可信数据交换、去中心化身份这类方向的 Python 服务，大概率会撞见affinidi-tdk-common这个名字。它是 Affinidi Trust Development Kit（TDK）里最底层的公共组件包，名字里的 “common” 不是随便叫的，它把整个 TDK 生态里所有服务都要用到的公共能力集中到了一起——配置读取、令牌获取与刷新、HTTP 客户端封装、统一异常、日志和序列化。

我第一次用这个包，是在一个用户数据存储后端里。当时项目已经跑了大半年，代码里散落着各种“自己手写的调用逻辑”：有人用requests直接打网关，有人复制了一段令牌缓存的代码，有人甚至连基础 URL 都是硬编码的。结果只要换一个环境、换一组密钥，就得在十几个文件里来回改。这种状态一长，我意识到问题的本质不是“某个接口写错了”，而是缺少一个公共层来统一处理那些所有服务都会面临的脏活累活。affinidi-tdk-common恰好就是干这个的。

这篇文章主要面向三类读者：一是正在用 Python 接入 Affinidi 生态、想快速上手的后端工程师；二是做内部工具、希望批量操作 Vault、Wallet 等服务的人；三是想看看企业级 Python 包是怎么设计配置、鉴权、重试和错误处理的人。我会从最基本的语法讲到参数含义，再给一个完整的实际应用案例，最后把我踩过的坑一并列出。

需要提前说明一点：这类包在持续迭代，不同小版本的方法命名可能略有差异。我下面以我常用的 v1.x 接口形态为例，更核心的是讲解思路和排查方法，你手里的版本就算类名有些出入，整体原则也完全通用。

2. 模块划分与设计思路拆解

2.1 不是所有代码都堆在__init__.py里

我见过很多工具包，为了省事把所有东西塞进一个巨大的__init__.py，最终变成几千行的“山”。一个真正可维护的公共库，一定会把职责拆分清楚。affinidi-tdk-common在常见设计里，基本围绕这几个模块来组织：

• config 模块：负责读取环境变量、配置文件，把各种来源的参数归一化成一个统一的配置对象，比如TDKConfig。
• auth 模块：负责和网关交换访问令牌，内部维护令牌缓存，并在令牌即将过期时自动刷新。
• http client 模块：封装底层 HTTP 请求，统一注入鉴权头、设置超时和重试策略，同时把网络异常翻译成业务异常。
• models 模块：定义请求和响应的数据模型，通常基于 Pydantic，做参数校验和字段序列化。
• errors 模块：定义统一异常类型，例如认证失败、请求超时、参数校验失败等。

这样分层的好处很直接：上层业务代码只需要面向高层 API 编程，不需要关心令牌是怎么拿到的、HTTP 超时是怎么配置的。比如我只要创建一个TDKClient(config)，之后调client.vault.get_data(...)时，令牌已经在底层自动处理好了。

2.2 为什么单独抽一个“公共层”这么重要

单独抽公共层，不只是为了少写代码，更多是为了让系统在演进过程中不那么脆。

举一个实际例子。某次我们更新了网关的令牌接口地址，旧地址被废弃。因为我们把所有鉴权逻辑收敛在 auth 模块里，只改了BASE_URL和一个obtain_token方法，所有依赖这个公共层的服务就都跟着修好了。要是当初每个服务各自实现一套鉴权，光排查哪些地方写死了旧地址，就能耗掉半天。

另一个理由是测试友好。公共层在单元测试阶段可以用固定配置初始化，或者直接把底层的 HTTP 调用 mock 掉，不需要真的连外网。比如我想测试“保存数据时如果令牌过期，底层会自动重新获取”，只需要 mock 掉 auth 模块的刷新方法，模拟第一次返回 401、第二次返回 200，就能快速验证重试逻辑。这种可测试性，对于稳定交付非常关键。

3. 语法与参数详解

3.1 安装、导入和版本确认

安装很简单，直接使用 pip：

pip install affinidi-tdk-common

如果你想确认当前安装的版本，可以用：

pip show affinidi-tdk-common

版本信息里会显示版本号、依赖项等。我建议在项目里固定版本，比如在requirements.txt或pyproject.toml里写成affinidi-tdk-common>=1.0,<2.0，避免第三方做了破坏性升级后，代码在没有测试的情况下被误升上去。

导入时，Python 会处理包名中的连字符，实际用下划线：

from affinidi_tdk_common import TDKConfig, TDKClient from affinidi_tdk_common.errors import TDKAuthError, TDKRequestError

3.2 配置参数逐项拆解

创建TDKConfig是最常用的一步。这个对象的意义，是把“环境变量、配置文件、代码传参”统一成一份可校验的配置。我常用的参数大概有这些：

这里有一个很容易踩的问题：private_key_path和private_key不能同时都传，也不要同时都不传。如果两个都传，不同实现的包可能会有不同行为，有的直接报错，有的默默选择其中一个，哪个是“正确的”完全看文档。为了安全，我通常只在配置代码里用private_key_path，只有在从密钥管理平台拿字符串时才用private_key。

创建配置对象的代码大概长这样：

config = TDKConfig( project_id="proj_12ab", token_id="tok_9x", private_key_path="./affinidi_private.key", environment="production", timeout=10.0, max_retries=3, )

你会发现这里把environment和api_url同时提供了。就我经验来看，最好让api_url显式存在，哪怕它是从环境变量里读出来的，也不要完全依赖“环境名自动映射地址”。因为真实生产环境里，你经常要连一个内部代理，或者指向某条灰度链路，光靠环境名推地址往往会踩坑。

我建议把关键配置都放进环境变量。常见的环境变量命名方式如下：

export AFFINIDI_PROJECT_ID="proj_12ab" export AFFINIDI_TOKEN_ID="tok_9x" export AFFINIDI_PRIVATE_KEY_PATH="/etc/affinidi/private.key" export AFFINIDI_API_URL="https://api.example.com"

然后在代码里统一读取：

import os config = TDKConfig( project_id=os.getenv("AFFINIDI_PROJECT_ID"), token_id=os.getenv("AFFINIDI_TOKEN_ID"), private_key_path=os.getenv("AFFINIDI_PRIVATE_KEY_PATH"), api_url=os.getenv("AFFINIDI_API_URL"), )

这种做法能避免隐私信息被写进代码仓库。注意，private_key_path对应的私钥文件绝对不要提交到 Git，要加入.gitignore，或者使用密钥管理系统在程序启动时注入到内存。

3.3 客户端对象与核心方法参数

拿到配置后，下一步是创建TDKClient：

client = TDKClient(config)

客户端内部会完成两件事：初始化认证模块，以及构建不同服务的调用入口。以 Vault 数据存取为例，我常用的调用方式是这样的：

resp = client.vault.save_data( data_id="user:001", items={ "name": "Alice", "level": 3, "tags": ["vip"], }, )

data_id是这条数据在 Vault 里的唯一标识，类似主键。items是你要存的实际内容，通常是一个 JSON 对象。这里最让我困惑的，是返回结果里到底哪些字段是稳定可依赖的。一般save_data的返回结构会包含data_id、version_id、created_at这些信息。version_id很有用，你可以把它理解为这条记录的第几次版本，后续如果要“只更新某个版本”或者做变更审查，它可以作为重要凭据。

读取数据的接口也很直观：

data = client.vault.get_data(data_id="user:001") print(data.items)

get_data返回的对象，一般会有一个字段来承载完整数据内容。实际项目中，我用 Pydantic 模型来承接返回内容，避免在业务代码里到处用dict取 key：

from pydantic import BaseModel from datetime import datetime class VaultRecord(BaseModel): data_id: str version_id: str created_at: datetime items: dict parsed = VaultRecord.model_validate(data.model_dump()) print(parsed.items["name"])

3.4 异常体系与错误处理语法

我最喜欢这个包的一点，是它把异常分成几个语义明确的类型。常见的包括：

• TDKAuthError：认证失败，比如私钥错误、令牌过期且刷新失败。
• TDKValidationError：请求参数不合法，比如data_id为空、items不是对象。
• TDKRequestError：网络请求层面的错误，比如超时、连接失败、服务返回 5xx。
• TDKApiError：业务接口返回错误码时的统一异常。

在业务代码里的处理就像这样：

try: result = client.vault.save_data( data_id="user:001", items={"name": "Alice"}, ) except TDKAuthError as e: # 密钥或令牌配置有问题，这里应该报警而不是重试 logger.error(f"auth failed: {e}") raise except TDKValidationError as e: # 参数写错了，属于程序 bug，也不该重试 logger.error(f"params invalid: {e}") raise except TDKRequestError as e: # 网络或下游问题，可以重试 logger.warning(f"request error: {e}, will retry") retry()

很多新手在没搞清异常类型时，喜欢“一把梭”地except Exception，这在本地脚本里能跑，但到生产就非常痛苦，因为你没办法区分哪些错误值得重试，哪些错误根本不值得浪费资源。这里的原则是：参数错误和认证错误不要重试，网络超时、连接失败、5xx 才值得重试。

4. 实操过程与核心环节实现

4.1 一个完整案例：用户数据可信存证服务

我想用一个非常贴近工作的场景来演示整套用法。假设你有一个后端服务，需要把用户的原始行为数据加密保存起来，并保留历史版本，方便后续审计。这个服务对外暴露 HTTP API，底层使用 Vault 来存储数据。

整个落地分四步走：

1. 安装依赖，确认版本。
2. 准备配置文件和私钥，从环境变量读取。
3. 初始化TDKClient，封装成可被 FastAPI 依赖注入的服务。
4. 编写核心业务逻辑，处理保存和读取。

4.2 初始化配置和客户端

首先在项目入口处加载配置。我会单独写一个dependencies.py，把所有初始化逻辑集中到这里：

import os from fastapi import FastAPI, Request from affinidi_tdk_common import TDKConfig, TDKClient def build_config() -> TDKConfig: return TDKConfig( project_id=os.getenv("AFFINIDI_PROJECT_ID"), token_id=os.getenv("AFFINIDI_TOKEN_ID"), private_key_path=os.getenv("AFFINIDI_PRIVATE_KEY_PATH"), api_url=os.getenv("AFFINIDI_API_URL", "https://api.example.com"), timeout=float(os.getenv("AFFINIDI_TIMEOUT", "10")), max_retries=int(os.getenv("AFFINIDI_MAX_RETRIES", "3")), ) def build_client() -> TDKClient: return TDKClient(build_config())

然后用 FastAPI 的 lifespan 特性初始化客户端，挂到app.state上：

from contextlib import asynccontextmanager from fastapi import FastAPI @asynccontextmanager async def lifespan(app: FastAPI): app.state.tdk = build_client() yield app.state.tdk.close() app = FastAPI(title="data-notary", lifespan=lifespan)

有人可能会问，每次请求都新建客户端不行吗？行，但不推荐。因为TDKClient内部会缓存令牌，如果每次请求都重建，等于抛弃了令牌缓存，每次都要重新走一遍拿令牌的流程，既慢又容易触发网关限流。客户端保持一个长期实例，才是合理的用法。

4.3 核心业务逻辑：保存与读取

下面写两个接口。第一个是写入数据：

from fastapi import Request, HTTPException from pydantic import BaseModel class SaveRequest(BaseModel): data_id: str payload: dict @app.post("/records") def save_record(req: SaveRequest, request: Request): client: TDKClient = request.app.state.tdk try: resp = client.vault.save_data( data_id=req.data_id, items=req.payload, ) return {"data_id": resp.data_id, "version_id": resp.version_id} except TDKAuthError as exc: raise HTTPException(status_code=401, detail=str(exc)) except TDKValidationError as exc: raise HTTPException(status_code=400, detail=str(exc)) except TDKRequestError as exc: raise HTTPException(status_code=502, detail=str(exc))

第二个是读取数据：

@app.get("/records/{data_id}") def get_record(data_id: str, request: Request): client: TDKClient = request.app.state.tdk try: data = client.vault.get_data(data_id=data_id) return {"data_id": data.data_id, "items": data.items} except TDKRequestError as exc: raise HTTPException(status_code=404, detail=f"data not found: {exc}")

注意，我在接口层捕获了公共层抛出的异常，并转换成 HTTP 状态码。这个设计很关键，因为底层 SDK 的异常模型，一般来说不应该直接暴露给前端或调用方。统一转成 HTTP 语义后，上游调用方才能明白发生了什么。

4.4 测试时的 Mock 与验证

这种公共层包的优势在测试时会特别明显。我可以把TDKClient整体 mock 掉，不需要真实连接任何服务：

from unittest.mock import Mock, patch def test_save_record_success(): fake_resp = Mock() fake_resp.data_id = "user:001" fake_resp.version_id = "v1" fake_client = Mock() fake_client.vault.save_data.return_value = fake_resp with patch("app.dependencies.build_client", return_value=fake_client): # 接下来走 FastAPI TestClient，验证接口返回 200 ...

如果我想更精细地验证“底层是否用正确的参数调用了接口”，可以用assert_called_once_with：

fake_client.vault.save_data.assert_called_once_with( data_id="user:001", items={"name": "Alice"}, )

如果不想 mock 整个客户端，只想 mock 底层 HTTP 调用，也可以用responses这类库拦截网络请求。但总体来说，mockTDKClient是最快、最稳的做法，因为公共层的内部实现不是我们测试的目标，我们测试的是业务接口的逻辑。

5. 常见问题排查与避坑实录

我在实际使用中积累了一些问题，下面按优先级排一下，都是踩过之后才明白的。

5.1 401 和 403：凭证相关的坑

最常见的 401 原因有两个：一是私钥和token_id不匹配，二是本地时间不准，导致签名校验窗口异常。

第一个原因好理解，生成token_id时绑定的公钥和当前私钥对不上，网关校验签名失败。排查时我可以先用官方提供的验证脚本跑一遍，确认凭证本身有效，再去看代码。

第二个原因很隐蔽。很多签名算法依赖时间戳，客户端生成请求时用本地时间，如果服务器时间偏差过大，签名会被判定为无效。我遇到过一台没有做时间同步的机器，时间慢了五分钟，所有请求都神秘失败。排查方法是打印请求签名时间和服务端返回错误里的时间差。解决方案也很简单，把 NTP 时间同步做好。

403 则更可能是权限问题。比如这个token_id只有读取权限，却调用了写接口。这种情况下，错误信息一般会明确指出缺少哪个权限。

5.2 令牌自动刷新失败

公共层一般会帮忙自动刷新令牌，但刷新失败的情况比想象中多。一个经典原因是，刷新操作本身用的还是同一个token_id和私钥，如果私钥已经轮换，但配置里的private_key_path还没更新，刷新就会持续失败。

处理方式是：把配置来源统一，不要在多个地方硬编码私钥。私钥轮换时，只改环境变量指向的新文件路径，而不是改动代码。同时日志里要有清晰的警戒日志，一旦刷新失败，立刻能定位到是配置过期而不是代码 bug。

5.3 JSON 序列化和字段不一致

Python 的datetime字段在序列化时容易出问题。当你拿到一个 Vault 记录，想把它返回给前端时，如果直接返回原始对象，可能会遇到“Object of type datetime is not JSON serializable”的错误。解决方法是使用 Pydantic 模型，或者手动把datetime转成 ISO 字符串。

另外，接口返回字段在不同版本里可能改名。比如旧版本返回created_at，新版本可能加了一个updated_at，但你的代码仍只读取旧字段。建议在模型层做一次字段映射，避免业务代码大量使用原始字典 key。

5.4 超时设置和重试风暴

很多人忽略timeout参数，导致在网关慢的时候，请求一直挂着不返回。特别是同步代码里，一个请求挂几十秒，整个线程池就被拖垮。我为TDKConfig里的timeout设置了至少 10 秒，并在调用链关键位置设置更短的连接超时。

重试也要谨慎。max_retries=3看起来无害，但如果所有请求在服务异常时同时触发重试，下游很容易被打爆。比较好的策略是：只对“幂等”的请求开启重试，例如读取数据、删除数据，而写操作要谨慎，避免重复写入造成数据污染。

下面把常见问题整理成速查表：

6. 一点个人经验

最后分享一个从多次踩坑中得出的体会：使用这类公共库时，一定不要绕过它，去直接调用底层 HTTP 接口。表面上看，绕过去能“更灵活”，比如自己拼一个带鉴权头的请求，但长期来看，你会丢掉令牌刷新、统一异常、重试策略这些现成能力。等某天网关升级或安全要求变化，你就得自己扛下所有底层细节。

我现在的习惯是，在项目启动阶段就把affinidi-tdk-common的配置封装成独立模块，并写几个简单的单元测试固定住关键行为。这种做法在初期多花半小时，但后面每次联调、排查问题时都能省回好几倍时间。如果你刚开始接触这个包，我建议你按文章里的结构先搭一个最小可运行的服务，再逐步加业务逻辑，跑通一遍之后，很多细节自然就理解了。