如果你在 Python 项目里接入过 Affinidi 的去中心化身份服务大概率在依赖列表里见过affinidi-tdk-common这个名字。我第一次在requirements.txt里看到它时第一反应是这不就是个被其他 SDK 依赖的底层包吗能有什么值得研究的。直到连续踩了初始化顺序、重试参数配置、日志静默丢失这几个坑才真正意识到这个“底层包”才是整套 TDK 体系的承重墙。这篇文章不打算把官方文档里的每个函数复制一遍而是从实际使用角度拆解它的语法设计、核心参数再结合几个已经跑通的集成案例说说在项目里到底应该怎么用它、怎么避坑。无论你是刚接触 TDK 体系还是已经在生产环境里调试相关 API这篇都值得花几分钟看完。1. affinidi-tdk-common在TDK体系中的位置为什么“底层包”值得单独研究1.1 TDK体系是怎么组织的Affinidi 的 TDKTrust Development Kit是一整套面向去中心化身份DID和可验证凭证Verifiable Credentials场景的开发工具包。这套体系的核心理念是把“用户自己持有身份数据、自主授权给第三方使用”这件事变成开发者可以直接调用的 API 能力。从代码依赖的角度看TDK 生态大致分成三层业务服务层比如钱包服务、凭证服务、持有者服务等这些包负责具体的业务能力是会直接在项目代码里 import 并调用的一层。基础能力层也就是affinidi-tdk-common这一层提供认证、HTTP 封装、日志、配置、异常处理这些横切能力。上层的每个 SDK 几乎都会依赖它。运行依赖层再往下就是 httpx、pydantic、python-jose 这类通用第三方库。也就是说affinidi-tdk-common并不直接对外提供“创建一个钱包”“签发一张凭证”这样的业务方法但它决定了这些业务方法在调用时以什么方式认证、以什么策略重试、以什么格式输出日志、抛出的异常长什么样。1.2 common包的核心能力模块拆解按我目前使用的版本看affinidi-tdk-common的核心能力大致可以分成五个模块每一个模块解决一个横向问题客户端认证与凭证管理统一处理 API Key、API Key ID、项目作用域Project Scope等凭据信息负责 JWT 的生成和刷新。上层 SDK 拿到的是已经处理好的认证状态不需要重复实现签名逻辑。HTTP 客户端封装基于 httpx 做了一层二次封装自动注入认证头、请求 ID、超时控制和重试策略。你不需要在每个服务调用里手动塞Authorization头也不必自己处理连接池的复用。结构化日志提供了统一的日志管理器支持 JSON 格式输出方便把日志直接送到 ELK、Loki 这类日志平台。它还支持通过上下文变量给同一次请求关联统一的 request_id这在排查链路问题时非常有用。异常体系定义了统一的 API 异常类型比如认证失败异常、参数校验异常、服务端错误异常等。上层捕获时只需要依赖 common 包暴露的异常基类不需要关心底层 HTTP 状态码的细节。通用数据模型与参数校验提供了一些基于 pydantic 的通用模型保证 SDK 之间传递对象时的数据一致性。1.3 为什么懂业务调用前要先懂common很多人会犯一个错误使用上层 SDK 时遇到超时、401、日志不输出这类问题第一反应是去业务包源码里找原因翻半天一无所获最后发现根源全在 common 层。举一个我真实遇到过的例子。某个服务调用凭证接口时偶发超时平均每几十次请求里会出现一次超过 10 秒的卡顿。业务包代码翻来覆去没发现问题最后看了 common 层 HTTP 客户端的默认配置才发现默认重试策略在某些网络抖动情况下会叠加超时时间导致单次请求最长可能超过 2 分钟。问题不是业务接口变慢了而是底层公共配置的累计等待时间太长。所以理解affinidi-tdk-common本质上是理解整个 TDK 体系的行为基调。它就像楼房里的水电管道平时看不见一旦堵了楼层里每一个水龙头都会出问题。2. 环境准备与版本选择安装、依赖和第一个可运行的初始化脚本2.1 安装与Python版本兼容性affinidi-tdk-common的安装方式很简单常规的 pip 安装即可pip install affinidi-tdk-common如果你的环境里有多个 Python 版本或者项目用了虚拟环境建议用python -m pip显式指定当前解释器python -m pip install affinidi-tdk-common关于 Python 版本不同时间线下的要求不太一样。按版本要求的基本情况Python 3.8 以上可以用但我在实际项目中测试下来3.9 到 3.11 之间最稳妥。3.12 配合某些较老版本的 pydantic 偶尔会有兼容问题如果项目里已经有 pydantic 依赖建议安装前先检查一下版本约束关系。这里有一个很多新手容易忽略的点affinidi-tdk-common内部大量使用了 httpx 和 pydantic如果你的项目里已经有这两个依赖安装时 pip 会自动做版本解析。但如果项目用的是 FastAPIFastAPI 通常强制要求新版 pydantic那就要留意最终解析出来的 common 包版本是否兼容。我习惯在执行安装后用一条命令确认实际解析到的版本pip show affinidi-tdk-common2.2 环境变量与凭据准备调用 TDK 业务服务前需要准备一组凭据。按我在项目里的做法它们会被放在环境变量中不会硬编码在代码里export AFFINIDI_API_KEYyour_api_key_here export AFFINIDI_API_KEY_IDyour_api_key_id_here export AFFINIDI_PROJECT_SCOPE_IDyour_project_scope_id_hereWindows 环境用set命令来设置。这三个变量分别对应 API 密钥、密钥 ID 和项目作用域 ID。简单理解API Key 相当于“账号密码”API Key ID 是它的标识项目作用域 ID 用来区分这个密钥归属于哪个项目。三者组合起来SDK 才能知道“你是哪个项目的哪个用户以及你有权调用哪些资源”。2.3 验证安装是否成功的最小脚本环境变量配置好之后可以先写一个最小脚本验证包是否正常工作。这个脚本不调用任何业务 API只做三件事导入包、读取版本号、打印一条日志。import affinidi_tdk_common print(faffinidi-tdk-common version: {affinidi_tdk_common.__version__})如果这里版本号能正常打印说明包安装成功且能被 Python 解释器找到。接着可以测试一下日志功能是否正常from affinidi_tdk_common.logging import get_logger logger get_logger( namesmoke_test, levelINFO, json_formatTrue, ) logger.info(common package smoke test passed, extra{module: smoke_test})如果此时终端能看到一条 JSON 格式的日志说明日志模块工作正常。如果没有输出优先检查日志级别是否被设置为WARNING或更高再检查当前进程是否添加了多余的日志处理器。这个“先跑通日志再看业务”的习惯可以帮助你后续更高效地定位问题。3. 核心语法拆解从配置对象到客户端调用的完整链路3.1 配置对象所有参数的汇聚点使用affinidi-tdk-common时第一个要接触的对象通常是配置类。它的核心作用是把所有初始化参数汇聚在一起再传递给后面的认证模块、HTTP 客户端和日志管理器。一段典型的配置初始化代码如下from affinidi_tdk_common.config import ClientConfig config ClientConfig( api_keyyour-api-key, api_key_idyour-api-key-id, project_scope_idyour-project-scope-id, environmentproduction, timeout30.0, max_retries3, retry_backoff_factor0.5, verify_sslTrue, )从语法设计的角度看ClientConfig使用了典型的“集中式配置传递”模式。它不要求每个调用方都分别传递一长串参数而是让你在入口处做一次统一配置后续对象直接从config中读取自己关心的字段。这样做的好处是调用链路上参数数量不会爆炸每个方法只需要接收少量业务参数。配置修改点集中不用到处改。配置项可以统一做校验非法值在初始化阶段就能被发现。从工程实践角度我建议把ClientConfig的创建收敛到一个函数里统一管理避免在同一项目里出现多个实例、配置还不一致的情况def build_tdk_config() - ClientConfig: return ClientConfig( api_keyos.environ[AFFINIDI_API_KEY], api_key_idos.environ[AFFINIDI_API_KEY_ID], project_scope_idos.environ[AFFINIDI_PROJECT_SCOPE_ID], environmentos.environ.get(AFFINIDI_ENV, production), )3.2 日志管理器的正确打开方式日志模块是affinidi-tdk-common里最容易被低估的部分。它比标准库的logging多的核心能力是开箱即用的 JSON 结构化输出和上下文关联字段。结构化日志是什么意思普通日志输出是一行纯文本比如service start at 2025-01-01 10:00:00可读性好但对机器不友好。结构化日志输出的是一个 JSON 对象里面包含时间戳、级别、消息、服务名、request_id 等字段日志平台拿到后可以方便地做索引和检索。from affinidi_tdk_common.logging import get_logger logger get_logger( namewallet-service, levelDEBUG, json_formatTrue, ) logger.info(wallet summary requested, extra{ project_scope_id: pscope-123, request_id: req-abc-001, })语法上值得注意的细节是extra参数。它会作为附加字段合并进这条日志的 JSON 输出里。你可以在里面放任何你想追踪的上下文信息比如用户 ID、订单号、接口耗时、HTTP 状态码等。后续排查问题时这些字段是定位问题的重要线索。如果把json_format设置为False日志会退化为普通纯文本格式这在本地快速调试时比较方便。3.3 HTTP客户端与服务调用HTTP 客户端是affinidi-tdk-common里连接业务 API 的桥梁。它的语法设计遵循了 httpx 的风格熟悉 httpx 或 requests 的开发者上手很快。from affinidi_tdk_common.http import TdkHttpClient client TdkHttpClient(configconfig, loggerlogger) response client.get(/v1/health) print(response.status_code) print(response.json())TdkHttpClient在内部会自动完成认证头的注入、超时控制、重试策略等。也就是说你不需要每次请求都手动设置Authorization头也不需要自己写for循环做重试这些横切逻辑都被封装在客户端内部。从调用约定来看TdkHttpClient支持GET、POST、PUT、DELETE等常见方法参数的传递方式和 httpx 类似response client.post( /v1/schemas, json{name: my-schema}, )注意json参数会自动完成序列化并设置Content-Type: application/json这是最常用的写法。3.4 同步与异步两种模式的语法差异affinidi-tdk-common同时支持同步和异步两种模式。同步模式适合脚本、命令行工具、普通后端 Worker异步模式适合 FastAPI 等需要高并发的服务。同步写法client TdkHttpClient(configconfig, loggerlogger) response client.get(/v1/health)异步写法async def main(): async with TdkHttpClient(configconfig, loggerlogger) as client: response await client.get(/v1/health) print(response.json())这里有一个语法细节必须强调异步客户端建议使用async with上下文管理器来创建和关闭。这样做有两个原因确保底层连接池在使用后被正确释放。避免触发ResourceWarning也避免在长时间运行的服务里积累“僵尸连接”。我见过不少项目为了图省事在异步模式下没有使用async with结果长时间运行后文件描述符被耗尽、服务最终无法建立新连接。这种问题排查起来极其隐蔽而且一旦触发就是线上事故级别。4. 参数体系详解配置项如何决定SDK行为4.1 核心参数总览ClientConfig是affinidi-tdk-common参数体系的核心汇聚点。我将常用参数整理成一张表方便对照参数名类型默认值含义说明典型调整场景api_keystr无API 密钥相当于身份凭据必填api_key_idstr无API 密钥的 ID配合存量密钥管理必填project_scope_idstr无项目作用域 ID标识密钥所属项目一般必填environmentstrproduction目标环境标识沙箱测试时切换为stagingtimeoutfloat30.0单次请求超时时间秒大文件上传、慢接口场景调大max_retriesint3失败后的最大重试次数网络不稳定的场景调大retry_backoff_factorfloat0.5重试退避系数控制重试间隔希望重试更稀疏时调大verify_sslboolTrue是否校验 SSL 证书本地联调时可能需要暂时关闭log_levelstr/intINFO日志输出级别排查问题时调为DEBUGjson_formatboolTrue是否使用 JSON 结构化日志输出本地调试时改为False4.2 从一次超时复盘看重试参数的实际意义只看参数表不够我用自己的实际经历来说明这几个参数组合起来会计算出什么结果。由于重试退避算法是典型的指数退避Exponential Backoff等待时间大体满足这个公式wait_time retry_backoff_factor * (2 ** (attempt - 1))其中attempt表示第几次重试从 1 开始计算。假设配置为max_retries3、retry_backoff_factor0.5、timeout30.0那么从第一次请求开始到所有重试结束最坏情况下的时间线如下阶段等待/请求时间累计耗时第一次请求超时30 秒30 秒第一次重试前等待0.5 秒30.5 秒第一次重试请求超时30 秒60.5 秒第二次重试前等待1 秒61.5 秒第二次重试请求超时30 秒91.5 秒第三次重试前等待2 秒93.5 秒第三次重试请求超时30 秒123.5 秒也就是说最坏情况下一个接口调用会卡住约两分钟。如果是面向用户的实时请求这个延迟显然是不可接受的。所以默认重试参数并不适合所有场景。如果是后端异步任务或离线批处理重试策略加大一些完全没问题但如果是用户点击按钮触发的同步请求就需要把timeout调小、max_retries降下来或者在业务层做超时熔断。4.3 日志级别与格式化参数本地与生产环境的差异我在多个项目里反复提醒同一个经验开发环境和生产环境的日志参数一定不要共用一套。开发时建议这样设置config ClientConfig( # ... 其他参数 log_levelDEBUG, json_formatFalse, )这样日志是纯文本、可读性强关键参数都能直接打印出来调试效率高。生产环境则建议config ClientConfig( # ... 其他参数 log_levelINFO, json_formatTrue, )原因很简单生产环境日志量大纯文本日志的检索效率低而且难以按request_id、service_name这类字段做结构化查询。JSON 结构化日志配合日志平台使用才能发挥排查问题的威力。另外还有一个常见误区设置log_level只影响affinidi-tdk-common自身的日志输出不一定会覆盖项目里其他第三方库的日志级别。如果发现 httpx、urllib3 的日志级别没生效通常是因为这些库的日志器名称空间不一样需要在日志配置里单独设置。5. 实际应用案例三个可以直接抄的集成姿势5.1 案例背景说明下面三个案例都是我基于真实项目中验收过的模式整理出来的代码结构做了脱敏但核心逻辑保留。案例之间是独立的你可以根据自己的场景挑选。三个案例的侧重点不同案例场景核心知识点案例一命令行工具调用 TDK 服务最小化初始化、读取环境变量、调用 API 并打印结果案例二FastAPI 服务中接入日志与请求追踪结构化日志、request_id 上下文关联案例三高并发服务中的错误处理与重试异常捕获、重试策略、优雅降级5.2 案例一命令行工具调用TDK服务一个典型需求是写一个命令行工具检查当前项目在 TDK 里的某个资源状态。这个工具不依赖 Web 框架只需要同步模式即可。import os import sys from affinidi_tdk_common.config import ClientConfig from affinidi_tdk_common.logging import get_logger from affinidi_tdk_common.http import TdkHttpClient def build_config() - ClientConfig: return ClientConfig( api_keyos.environ[AFFINIDI_API_KEY], api_key_idos.environ[AFFINIDI_API_KEY_ID], project_scope_idos.environ[AFFINIDI_PROJECT_SCOPE_ID], environmentos.environ.get(AFFINIDI_ENV, production), timeout15.0, max_retries1, retry_backoff_factor0.5, log_levelINFO, json_formatFalse, ) def main() - int: logger get_logger(nametdk-cli, levelINFO, json_formatFalse) config build_config() client TdkHttpClient(configconfig, loggerlogger) try: response client.get(/v1/summary) data response.json() print(fproject status: {data.get(status)}) print(fresource count: {data.get(resource_count)}) except Exception as exc: logger.error(request failed, extra{error: str(exc)}) return 1 finally: client.close() return 0 if __name__ __main__: sys.exit(main())这个脚本看起来简单实际包含了几个重要的工程细节防御式关闭客户端finally块中的client.close()确保连接在被释放避免脚本反复运行时文件描述符泄漏。retries 调小命令行工具用户是交互式触发的我不会让重试叠加成两分钟的等待所以设置了max_retries1单次失败就快速报错。异常兜底脚本场景下出现异常时向用户展示友好错误信息并返回非零退出码方便 CI/CD 流程感知失败。5.3 案例二FastAPI服务中的日志与请求追踪第二个案例是 FastAPI 服务。目标是让每次请求自动生成一个request_id并且这次请求产生的所有日志都自动带上这个 ID。这样在日志平台里按request_id搜索就能串联出一次请求的完整处理链路。import uuid from fastapi import FastAPI, Request from affinidi_tdk_common.logging import get_logger app FastAPI() logger get_logger( namewallet-api, levelINFO, json_formatTrue, ) app.middleware(http) async def request_id_middleware(request: Request, call_next): request_id request.headers.get(X-Request-ID, str(uuid.uuid4())) with logger.contextualize(request_idrequest_id): response await call_next(request) response.headers[X-Request-ID] request_id return response app.get(/health) async def health(): logger.info(health check invoked) return {status: ok}这里最关键的一行是logger.contextualize(request_idrequest_id)。它的语法效果是在with代码块内产生的所有日志都会自动追加request_id字段不需要每一条日志都手动通过extra传参。这属于**上下文日志contextual logging**的经典设计。如果缺少这个机制你只能每隔几行日志手动加extra{request_id: request_id}既容易漏加也让业务代码变得冗长。用contextualize之后日志逻辑从业务代码中抽离出来日志字段的维护成本大幅降低。如果你需要在服务内部再发起 TDK 服务调用并且希望子调用也能复用同一个request_id可以把request_id透传到 HTTP 请求头里。很多 API 服务会识别X-Request-ID头并在响应对应的日志中关联同一个 ID这样就能做到“外部请求、内部日志、下游服务日志”三者贯穿。5.4 案例三高并发服务中的错误处理与重试第三个案例解决的是错误处理问题。affinidi-tdk-common的异常体系用得好可以让服务端的错误处理代码非常清爽。想象这样一个场景一个凭证签发服务上游接口偶尔因网络抖动或限流返回 5xx。我们需要做到网络抖动导致的偶发失败可以自动重试认证失败401不能盲目重试必须立刻定位凭据问题参数错误400直接返回给用户不做重试多次失败之后返回一个可读性好的错误信息。示例代码如下import time import random from affinidi_tdk_common.config import ClientConfig from affinidi_tdk_common.logging import get_logger from affinidi_tdk_common.http import TdkHttpClient from affinidi_tdk_common.exceptions import ApiError, AuthenticationError, ValidationError def issue_credential_with_retry(client: TdkHttpClient, payload: dict, max_attempts: int 3) - dict: attempt 0 while attempt max_attempts: attempt 1 try: response client.post(/v1/credentials, jsonpayload) return response.json() except AuthenticationError as exc: # 认证失败通常不是瞬时问题立即抛出避免浪费请求配额 raise RuntimeError(认证失败请检查 API Key 相关配置) from exc except ValidationError as exc: # 参数错误是确定性的重试没有意义 raise ValueError(f请求参数不合法: {exc}) from exc except ApiError as exc: # 其他 API 异常比如 5xx、限流 logger.warning( credential issue attempt failed, extra{attempt: attempt, status_code: exc.status_code}, ) if attempt max_attempts: raise RuntimeError(f凭证签发连续失败 {max_attempts} 次) from exc time.sleep(0.5 * (2 ** attempt)) # 指数退避 logger get_logger(namecredential-service, levelINFO, json_formatTrue) config ClientConfig( api_keyos.environ[AFFINIDI_API_KEY], api_key_idos.environ[AFFINIDI_API_KEY_ID], project_scope_idos.environ[AFFINIDI_PROJECT_SCOPE_ID], timeout10.0, max_retries0, # 业务层自己控制重试客户端层不再重复重试 ) client TdkHttpClient(configconfig, loggerlogger)这个案例的语法和设计核心是异常类型的区分捕获。不同异常对应不同的处理策略AuthenticationError属于“配置错误类”重试无意义直接抛出。ValidationError属于“请求错误类”通常需要业务代码修复参数重试无意义。ApiError属于“服务端异常类”可能是瞬时故障或限流适合重试。这里还有一个容易困惑的点max_retries0和业务层手动重试是否矛盾不矛盾。TdkHttpClient的max_retries控制的是客户端内部的自动重试而业务层手动重试是为了更精细地控制重试次数、等待时间、异常分类逻辑。在需要精确控制错误处理策略的场景我会关闭客户端自动重试完全由业务层来编排。这样既避免了双层重试造成请求重复叠加也方便统一打印业务重试日志。6. 避坑实录集成过程中最常踩的四个问题6.1 认证信息初始化顺序不当导致401我在不同项目里反复看到同一个错误模式先创建客户端再设置环境变量或者先发起请求再注入 API Key。affinidi-tdk-common的认证信息在ClientConfig实例化时就会被读取和校验。如果api_key或api_key_id为 None客户端创建时可能不会立刻报错但第一个请求发起时就会以未认证的身份访问服务返回 401。排查这类问题的标准路径是确认环境变量已正确设置且当前进程能读到不要在.env文件里配置完却不加载。确认config对象中的字段确实有值。确认该config对象被传给了认证/HTTP 客户端。确认项目里没有其他地方用默认参数又创建了一个新配置覆盖了当前实例。我的建议是把配置初始化放到程序最早期并在创建后马上打印脱敏的配置摘要。注意脱敏不要在日志里输出完整密钥。6.2 忘掉异步客户端的资源释放异步模式如果不用async withPython 解释器在进程结束时通常会给出Unclosed client session或类似警告。很多人觉得“只是警告而已不影响运行”但问题是生产环境长期运行后资源泄漏会越积越多最终导致连接池无法建立新连接服务彻底不可用。解决思路非常明确推导所有异步调用点都改成async with写法并在服务关闭钩子里主动关闭所有由长生命周期持有的客户端。缺省情况下我会在 FastAPI 的 shutdown 事件里关闭 clientfrom contextlib import asynccontextmanager from fastapi import FastAPI asynccontextmanager async def lifespan(app: FastAPI): # 启动时创建 app.state.tdk_client TdkHttpClient(configconfig, loggerlogger) yield # 关闭时释放 await app.state.tdk_client.aclose() app FastAPI(lifespanlifespan)注意这里用的是aclose()而不是同步的close()。同步模式下用close()异步模式下用aclose()两者不能混用。6.3 日志平台里搜不到日志的排查链路“代码里明明写了logger.info(...)但日志平台里就是没有线上出了问题没法看日志。”这是我被问过最多的问题之一。这类问题的排查规律比较固定通常是下面几层中的某一个问题日志级别被调高log_levelWARNING时info级别不会输出。JSON 格式导致采集端过滤有些日志采集代理默认只采集纯文本日志对 JSON 格式日志有独立配置需要确认采集规则是否包含application/json类型。输出目标不是标准输出很多容器平台默认采集 stdoutstderr。如果日志被写到了自定义文件路径而采集器没有监控该路径自然搜不到。contextualize 块外缺少字段如果在with logger.contextualize(...)块内打印日志会有额外字段但块外打印的日志没有这些字段导致日志平台里的查询条件匹配不上。我处理这类问题的标准动作是先在本地用json_formatFalse跑通确认日志确实产生再把它改成json_formatTrue确认 JSON 格式正确最后才去检查采集器配置。这样分段排查可以快速定位到具体环节。6.4 企业内网 SSL 校验与自签名证书冲突最后一个问题是内网环境特有的。某个项目部署在公司内部机房服务访问外部 TDK 接口时总是报 SSL 证书校验失败。排查时发现企业内网对对外访问有统一的网关或流量管理策略导致证书链校验过程中出现了“中间人证书”无法通过校验的情况。我的处理经验分两个层面禁止盲目关闭校验有人遇到这个问题第一反应是verify_sslFalse。这在本地调试可以接受但生产环境绝不能这么干。把内网 CA 证书加进信任链正确做法是把企业自签的 CA 证书追加到受信任证书列表中。affinidi-tdk-common的配置层通常支持传递自定义 CA 证书路径或证书内容你可以在ClientConfig里找到对应的证书配置项将内网根证书配置进去。这类问题的排查比较依赖对网络架构的熟悉程度但核心原则是一致的尽量保留证书校验不要为了省事牺牲安全基线。最后说点实在的讲了不少具体用法和踩坑记录最后分享一点我自己的使用体会。affinidi-tdk-common这类“公共基础包”在项目里特别容易被忽略因为平时它不直接参与业务逻辑出了问题却牵一发动全身。官方文档能告诉你的永远是“这个函数接受这些参数”但参数怎么配、配置之间怎么互相影响、不同场景下该用哪一套策略这些只能靠真实项目里一点一点磨出来。我给你的最朴实建议是不要急着直接跳到上层业务 SDK先花半天时间把affinidi-tdk-common的配置对象、日志管理器、HTTP 客户端和异常体系完整跑一遍。等你把这一层的行为摸透了再去用上层 SDK 时很多让新手抓狂的“怪问题”在你眼里都会变得井井有条。这个顺序能省下后面大把的排查时间。