iii-helpers Python 包 API 完全指南HTTP、OpenTelemetry 可观测性、Stream 与 RBAC 辅助类型【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iiiiii-helpers是 III 项目为 Python 开发者提供的官方辅助库它把引擎侧的基础设施能力封装成一组可直接 import 的类型与函数把引擎下发的原始 dict 请求包装成强类型StreamRequest/StreamResponse的http装饰器、以 OpenTelemetry LogRecord 为出口的结构化Logger、覆盖 traceparent/baggage 的追踪上下文工具、Stream 触发器的全套输入输出与原子更新操作set/merge/append等以及 Worker 连接管理所需的 RBAC 认证与注册钩子类型。读完本文你将能独立为 III Worker 编写带类型标注的 HTTP 处理器、接入链路追踪与结构化日志、操作 Stream 数据并实现基于 RBAC 的连接鉴权。本文以 docs/0-20-0/api-reference/helpers-python.mdx 为骨架并结合sdk/packages/python/helpers下的真实源码展开讲解。该文档由docs/next/scripts/generate-api-docs.mts自动生成正文源文件是sdk/packages/python/helpers/src各模块的 docstring因此源码注释与文档内容一一对应可直接对照阅读。安装与包结构安装只需一条命令pip install iii-helpers安装后可以从五个子模块导入所需内容与源码目录sdk/packages/python/helpers/src/iii_helpers/一一对应子模块用途源码位置iii_helpers.httpHTTP 请求/响应类型、认证配置与http辅助函数http/init.pyiii_helpers.observability结构化 Logger、OpenTelemetry 配置与 span 工具observability/init.pyiii_helpers.queue队列入队结果类型queue/init.pyiii_helpers.streamStream 触发器配置、变更事件、IO 输入输出与更新操作stream/init.pyiii_helpers.worker_connection_managerRBAC 认证与注册回调类型worker_connection_manager/init.pyhelpers/tests/下提供了test_observability_telemetry.py与test_observability_http_instrumentation.py两组测试sdk/packages/python/iii/tests/下还有test_stream_models.py、test_stream_types.py等测试可作为理解各类型行为边界的参考。HTTP 辅助模块iii_helpers.httpHTTP 模块解决的是引擎把原始请求数据交给 Python 处理器的适配问题函数内部无需手工解析 dict而是拿到带类型的请求/响应对象。http()把流式 handler 包装成引擎可直接调用的函数http接收一个回调(req, res) - HttpResponse | None返回一个 III 引擎可直接调用的异步函数。包装器负责把引擎下发的原始 dict或InternalHttpRequest转换成回调期望的强类型StreamRequest/StreamResponse对。签名http(callback: Callable[Awaitable[HttpResponse[Any] | None]])从源码看http的wrapper会依次处理三种输入形态见 http/init.py输入是InternalHttpRequest实例直接使用输入是dict从path_params、query_params、body、headers、method、response、request_body键还原出InternalHttpRequest其他类型原样透传。随后构造StreamResponse(internal.response)与StreamRequest(...)并调用用户回调。这意味着你只需关注类型化的HttpRequest视图底层的 WebSocket 响应通道由包装器代为维护。用法示例from iii_helpers.http import http, HttpResponse app.function(http_handler) http async def handler(req, res): # req: StreamRequest含 method / headers / path_params / query_params / body return HttpResponse(status_code200, body{ok: True})HttpInvocationConfigHTTP 外部函数调用配置该配置用于描述对外部 HTTP 服务如 Lambda、Cloudflare Workers的调用参数字段类型必填说明urlstr否调用的目标 URLmethodHttpMethod否HTTP 方法默认POSTtimeout_msint \| None否请求超时毫秒headersdict[str, str] \| None否附加请求头authHttpAuthConfig \| None否认证配置bearer / HMAC / API key其中HttpMethod是源码中定义的Literal[GET, POST, PUT, PATCH, DELETE]注意它与核心builtin_triggers的 HTTP 方法枚举不同后者还覆盖 HEAD/OPTIONS见 http/init.py。三种认证类型认证字段的值都指向环境变量名而不是明文密钥本身敏感信息不会出现在代码或配置里类型字段说明HttpAuthApiKeyheaderAPI key 的自定义请求头名value_key存放 API key 值的环境变量名API key 通过自定义请求头发送HttpAuthBearertoken_key存放 bearer token 的环境变量名Bearer token 认证HttpAuthHmacsecret_key存放 HMAC 共享密钥的环境变量名基于共享密钥的 HMAC 签名校验三个模型都有type判别字段Literal[api_key]/bearer/hmac并由HttpAuthConfig HttpAuthHmac | HttpAuthBearer | HttpAuthApiKey组成联合类型。HttpRequest与HttpResponseHttpRequest表示一个已缓冲的 HTTP 请求字段类型说明methodstr请求方法如GET、POSTpath_paramsdict[str, str]从匹配路由提取的路径参数query_paramsdict[str, str \| list[str]]URL 查询串参数headersdict[str, str \| list[str]]请求头bodyAny \| None已解析的请求体HttpResponse表示要返回的缓冲响应字段类型说明status_codeintHTTP 状态码headersdict[str, str]响应头bodyAny \| None响应体model_configAnyPydantic 模型配置值得注意的实现细节HttpResponse的 Pydantic 配置为ConfigDict(populate_by_nameTrue, arbitrary_types_allowedTrue)且status_code在序列化时使用别名statusCode与引擎侧和 Rust SDK 的线缆格式保持一致见 http/init.py。可观测性模块iii_helpers.observability该模块是 Python Worker 接入可观测性的核心涵盖结构化日志、OpenTelemetry 初始化与 span 操作。Logger以 OTel LogRecord 为出口的结构化日志器Logger把每条日志作为 OpenTelemetry LogRecord 发出每次调用都会自动捕获当前 trace 与 span 上下文让日志与分布式链路天然关联无需手工拼接 trace id。当 OTel 未初始化时Logger会优雅地回退到 Python 标准logging。源码实现见 observability/logger.py展示了两个关键设计四级方法info/warn/error/debug映射到 OTelSeverityNumberINFO9、WARN13、ERROR17、DEBUG5日志属性中自动携带service.name结构化数据以log.data属性整体附加trace_id / span_id 优先取显式构造参数否则取当前 span 上下文。推荐用法把结构化数据作为第二个参数传入from iii import Logger logger Logger() # 基础日志trace 上下文自动注入 logger.info(Worker connected) # 结构化上下文便于在可观测后端过滤、聚合、建仪表盘 logger.info(Order processed, {order_id: ord_123, amount: 49.99, currency: USD}) logger.warn(Retry attempt, {attempt: 3, max_retries: 5, endpoint: /api/charge}) logger.error(Payment failed, { order_id: ord_123, gateway: stripe, error_code: card_declined, })官方文档特别强调使用 dict 键值对而不是字符串插值才能在 Grafana、Datadog 等后端做过滤与聚合。另外Logger构造函数还接受可选的trace_id/span_id/service_name用于在缺少活动 span 时显式绑定上下文。OtelConfigOpenTelemetry 初始化配置init_otel(config: OtelConfig | None None, loop: None None)负责初始化 OpenTelemetry后续重复调用为 no-op。其配置字段源码见 observability/telemetry_types.py字段类型默认值 / 说明enabledbool \| None是否启用 OTel默认 True设OTEL_ENABLEDfalse/0/no/off可关闭service_namestr \| None服务名默认取环境变量OTEL_SERVICE_NAME否则iii-python-sdkservice_versionstr \| None服务版本默认取SERVICE_VERSION否则unknownservice_namespacestr \| None服务命名空间属性service_instance_idstr \| None服务实例 ID默认随机 UUIDengine_ws_urlstr \| NoneIII 引擎 WebSocket 地址默认取环境变量III_URL否则ws://localhost:49134fetch_instrumentation_enabledbool是否通过URLLibInstrumentor自动埋点 urllib HTTP 调用默认 Truespans_flush_interval_msint \| Nonespan 处理器刷盘延迟默认 100ms环境变量覆盖OTEL_SPANS_FLUSH_INTERVAL_MS。文档特别提示OpenTelemetry 默认的 5000ms 正是操作结束后几秒才看到 trace的原因logs_enabledbool \| None是否经EngineLogExporter导出 OTel 日志OTel 启用时默认 Truelogs_flush_interval_msint \| None日志处理器刷盘延迟默认 100mslogs_batch_sizeint \| None每批导出的日志条数上限默认 1metrics_enabledbool是否经EngineMetricsExporter导出指标默认 Truemetrics_export_interval_msint指标导出间隔毫秒默认 6000060 秒ReconnectionConfigWebSocket 重连策略字段类型默认值说明initial_delay_msint1000起始延迟毫秒max_delay_msint30000最大延迟上限毫秒backoff_multiplierfloat2.0指数退避乘数jitter_factorfloat0.3随机抖动因子0–1max_retriesint-1最大重试次数-1表示无限重试Span 工具函数函数签名要点行为current_trace_id()无参返回当前活动 trace_id 的 32 位十六进制字符串不可用时返回 Nonecurrent_span_id()无参返回当前活动 span_id 的 16 位十六进制字符串不可用时返回 Nonecurrent_span_is_recording()无参无活动 span 或采样器丢弃 span 时返回 Falsewith_span(name, fn, kindNone, traceparentNone)异步启动新 span 并在其中运行fn(span)tracer 未初始化时用 no-op span 调用fn静默忽略属性/事件调用set_current_span_attribute(key, value)同步给当前 span 设置属性span 未 recording 时为 no-oprecord_span_event(name, attrsNone)同步记录 span 事件span 未 recording 时为 no-opset_current_span_error(message)同步标记当前 span 错误无活动 span 时为 no-op追踪上下文注入与提取跨服务传播追踪信息时使用 W3C 标准头函数行为inject_traceparent()把当前 trace 上下文注入 W3Ctraceparent头字符串extract_traceparent(traceparent: str)从 W3Ctraceparent头字符串提取 trace 上下文inject_baggage()把当前 baggage 注入 W3Cbaggage头字符串extract_baggage(baggage: str)从 W3Cbaggage头字符串提取 baggage配套的BaggageSpanProcessor负责把 baggage 中的键值同步到 span 上保证跨服务调用时业务上下文不丢失。生命周期与带追踪的 HTTP 调用init_otel(configNone, loopNone)初始化 OpenTelemetry重复调用为 no-opshutdown_otel()同步关闭 OTel尽力而为不等待 WS 刷盘flush_otel()shutdown_otel的对立面——在不拆除提供者的前提下强制刷盘所有 OTel 提供者。适合在短生命周期进程退出前使用既希望挂起的 spans/metrics/logs 被送达又计划继续使用 OTelexecute_traced_request(client, request)在 OTel CLIENT span 内执行 httpx 请求。具体行为包括向出站请求头注入 W3Ctraceparent在 span 上记录 HTTP 语义约定属性对 status 400 的响应设置 ERROR span 状态对网络级错误记录异常。日志脱敏与截断可观测性数据经常包含敏感字段模块提供了脱敏工具函数说明redact(value: Any)对值进行脱敏处理redact_and_truncate(value: Any, max_bytes: Optional[int] None)脱敏并按字节上限截断resolve_max_bytes_from_env()从环境变量解析截断上限DEFAULT_ALLOWLISTtuple[str, ...]类型的默认放行键集合这些键不被脱敏队列模块iii_helpers.queueEnqueueResult是函数以TriggerAction.Enqueue方式被调用时返回的结果类型字段类型说明messageReceiptIdstr引擎为入队任务分配的 UUID 回执 IDfrom iii_helpers.queue import EnqueueResult # 在函数内把消息投递到队列后返回回执 return EnqueueResult(messageReceiptIdreceipt_id)Stream 模块iii_helpers.streamStream 是 III 中流式数据 状态的核心抽象该模块提供四类类型触发器配置、变更事件、IO 输入输出、更新操作。Stream 触发器配置StreamTriggerConfig用于配置stream触发器决定哪些条目变更会触发 handler字段类型必填说明stream_namestr是要监听的流名称只有该流上的变更触发 handlergroup_idstr \| None否设置后仅该组内的变更触发item_idstr \| None否设置后仅该条目的变更触发condition_function_idstr \| None否条件函数 ID返回 False 时跳过 handlerStreamJoinLeaveTriggerConfig用于stream:join/stream:leave触发器仅含condition_function_id一个可选字段。Stream 事件类型StreamChangeEvent是stream触发器的 handler 输入在条目通过stream::set/stream::update/stream::delete变更时触发字段类型必填说明typeLiteral[stream]是事件类型timestampint是事件的 Unix 时间戳streamNamestr是变更发生的流groupIdstr是变更发生的组idstr \| None否变更的条目 IDeventStreamChangeEventDetail是变更详情StreamChangeEventDetail包含type: Literal[create, update, delete]变更类型与data: Any关联数据。StreamJoinLeaveEvent表示流的加入/离开事件字段包括subscription_id唯一订阅标识、stream_name、group_id、id可选条目 ID与context来自StreamAuthResult的认证上下文。认证相关类型类型字段说明StreamAuthInputheaders: dict[str, str]、path: str、query_params: dict[str, list[str]]、addr: str均必填流认证输入StreamAuthResultcontext: Any \| None认证通过后传给 stream handler 的任意上下文StreamJoinResultunauthorized: bool必填加入是否未授权Stream IO 输入输出操作输入字段结果字段Getstream_name、group_id、item_id—Setstream_name、group_id、item_id、data: Anyold_value: TData \| None、new_value: TDataDeletestream_name、group_id、item_idold_value: Any \| NoneListstream_name、group_id—ListGroupsstream_name—Updatestream_name、group_id、item_id、ops: list[UpdateOp]原子应用的有序操作列表old_value、new_value、errors: list[UpdateOpError]更新操作Update OpsUpdateOp是六种操作的联合类型UpdateSet | UpdateIncrement | UpdateDecrement | UpdateAppend | UpdateRemove | UpdateMerge。操作类型标识字段语义UpdateSetsetpath: str、value: Any把路径字段设为指定值空字符串表示根值UpdateIncrementincrementpath: str、by: int \| float数值字段增加byUpdateDecrementdecrementpath: str、by: int \| float数值字段减少byUpdateRemoveremovepath: str删除路径字段UpdateAppendappendpath: MergePath \| None、value: Any数组追加 / 字符串拼接 / 嵌套路径 pushUpdateMergemergepath: MergePath \| None、value: Any把对象浅合并到目标节点MergePath str | list[str]即单字符串旧式/一级键或字面量段列表嵌套路径与 Node SDK 的string | string[]别名及 Rust 的MergePath枚举对齐。UpdateAppend的路径形式与引擎语义接受的路径形式与UpdateMerge一致None//[]在根节点追加foo在一级键foo处追加。注意带点的字符串如a.b是字面量键名a.b不会被拆解成a - b[a, b, c]嵌套路径每个元素都是字面量段。引擎在叶子节点的行为嵌套路径上的缺失/非对象中间节点自动以{}创建/替换叶子缺失/null 嵌套路径 →[value]总是数组叶子缺失/null 单字符串路径 → 字符串拼接层级按字符串处理否则[value]已存在数组 → push已存在字符串 字符串值 → 拼接叶子是对象/标量 → 返回append.type_mismatch。UpdateMerge的路径形式与引擎语义根合并None//[]foo等价于[foo]即一级键[a, b, c]嵌套路径每个元素是字面量键[a.b]写入的是名为a.b的单个键而不是a - b。引擎行为路径上的缺失/非对象中间节点自动替换为{}合并是目标节点处的浅合并value的顶层键覆盖同名键兄弟键保留。安全与合法性校验重点merge与append都会做严格的输入校验违规操作返回结构化错误且该操作不会生效路径深度 32 段段长度 256 字节值深度 16merge顶层键 1024 个merge任何__proto__/constructor/prototype段或顶层键防原型污染。错误通过state::update/stream::update响应的errors数组返回成功应用的操作仍反映在响应的new_value中。UpdateOpError每个失败操作的错误详情字段类型说明op_indexint出错操作在原始ops数组中的下标codestr稳定错误码如merge.path.too_deepmessagestr可读的错误描述含具体数值doc_urlstr \| None可选的文档链接实现细节UpdateAppend与UpdateMerge都通过model_serializer(modewrap)在序列化时省略path: None从而让线缆载荷与 Rust SDK 的#[serde(skip_serializing_if Option::is_none)]逐字节一致见 stream/init.py。StreamUpdateResult.errors为空时也会从 JSON 线缆中省略该字段。Worker 连接管理模块RBAC该模块为通过 RBAC 端口连接的 Worker 提供认证输入/输出与三类注册钩子的类型定义是构建多租户、细粒度权限控制的基础。AuthInput与AuthResultAuthInput是 WebSocket 升级时传给 RBAC 认证函数的输入包含升级请求的 HTTP 头、查询参数与客户端 IP字段类型说明headersdict[str, str]WebSocket 升级请求的 HTTP 头query_paramsdict[str, list[str]]升级 URL 的查询参数每个键对应值列表以支持重复键ip_addressstr连接客户端 IPAuthResult控制认证后的 Worker 可以调用哪些函数、注册哪些触发器以及向中间件转发什么上下文字段类型说明allowed_functionslist[str]在expose_functions之外额外允许的函数 IDforbidden_functionslist[str]即使匹配expose_functions也拒绝的函数 IDallowed_trigger_typeslist[str] \| None允许注册触发器的触发器类型 IDNone表示全部允许allow_trigger_type_registrationbool是否允许注册新的触发器类型allow_function_registrationbool是否允许注册新函数function_registration_prefixstr \| None应用于该 Worker 注册的所有函数 ID 的前缀contextdict[str, Any]每次调用转发给中间件的任意上下文从源码看见 worker_connection_manager/init.pyAuthResult还支持namespaces: dict[str, list[str]]字段用于按命名空间授予作用域化权限例如{orders: [svc::*]}值可以是精确函数 ID也可以是通配符支持裸写或match(...)写法。当namespaces为空时会话可声明任意命名空间仅受allowed_functions与expose_functions约束。注册钩子三类钩子分别在 Worker 通过 RBAC 端口注册函数、触发器、触发器类型时被调用返回映射后的结果或抛异常拒绝注册。省略的字段保留注册请求中的原始值。函数注册on_function_registration_function_idOnFunctionRegistrationInputfunction_id、description、metadata、context会话的认证上下文OnFunctionRegistrationResultfunction_id、description、metadata均为可选省略即保留原值。触发器注册on_trigger_registration_function_idOnTriggerRegistrationInputtrigger_id、trigger_type、function_id、config触发器专属配置、metadata、contextOnTriggerRegistrationResulttrigger_id、trigger_type、function_id、config均为可选映射结果。触发器类型注册on_trigger_type_registration_function_idOnTriggerTypeRegistrationInputtrigger_type_id、description、contextOnTriggerTypeRegistrationResulttrigger_type_id、description均为可选映射结果。从源码还可看到函数与触发器注册输入都包含namespace字段显式注册值缺省为default由于同一 ID 可以存在于多个命名空间钩子需要依据命名空间逐项授权。小结三类落地场景综合以上五个模块iii-helpers在 III 项目中的典型组合方式是HTTP 服务用http()包装 handler 获得强类型请求/响应配合HttpInvocationConfig与三种HttpAuth*配置调用外部 HTTP 服务密钥全部走环境变量可观测性用OtelConfig初始化 OTelLogger输出与 trace 关联的结构化日志with_span/set_current_span_attribute/execute_traced_request串联分布式调用flush_otel在短进程退出前保证数据送达状态与实时流用StreamTriggerConfig订阅变更通过StreamSetInput/StreamUpdateInput内含merge、append等原子操作与原型污染防护读写流数据并用worker_connection_manager的类型构建 RBAC 认证与注册策略把多租户权限落到函数、触发器与命名空间三个粒度上。如需继续深入可阅读同版本配套文档 docs/0-20-0/sdk-reference/ 下的 SDK 指南以及sdk/packages/python/iii/tests/下的test_stream_models.py、test_stream_types.py、test_rbac_workers.py等测试了解各类型在真实调用链中的行为。【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 iii-helpers Python 包 API 完全指南：HTTP、OpenTelemetry 可观测性、Stream 与 RBAC 辅助类型
iii-helpers Python 包 API 完全指南HTTP、OpenTelemetry 可观测性、Stream 与 RBAC 辅助类型【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iiiiii-helpers是 III 项目为 Python 开发者提供的官方辅助库它把引擎侧的基础设施能力封装成一组可直接 import 的类型与函数把引擎下发的原始 dict 请求包装成强类型StreamRequest/StreamResponse的http装饰器、以 OpenTelemetry LogRecord 为出口的结构化Logger、覆盖 traceparent/baggage 的追踪上下文工具、Stream 触发器的全套输入输出与原子更新操作set/merge/append等以及 Worker 连接管理所需的 RBAC 认证与注册钩子类型。读完本文你将能独立为 III Worker 编写带类型标注的 HTTP 处理器、接入链路追踪与结构化日志、操作 Stream 数据并实现基于 RBAC 的连接鉴权。本文以 docs/0-20-0/api-reference/helpers-python.mdx 为骨架并结合sdk/packages/python/helpers下的真实源码展开讲解。该文档由docs/next/scripts/generate-api-docs.mts自动生成正文源文件是sdk/packages/python/helpers/src各模块的 docstring因此源码注释与文档内容一一对应可直接对照阅读。安装与包结构安装只需一条命令pip install iii-helpers安装后可以从五个子模块导入所需内容与源码目录sdk/packages/python/helpers/src/iii_helpers/一一对应子模块用途源码位置iii_helpers.httpHTTP 请求/响应类型、认证配置与http辅助函数http/init.pyiii_helpers.observability结构化 Logger、OpenTelemetry 配置与 span 工具observability/init.pyiii_helpers.queue队列入队结果类型queue/init.pyiii_helpers.streamStream 触发器配置、变更事件、IO 输入输出与更新操作stream/init.pyiii_helpers.worker_connection_managerRBAC 认证与注册回调类型worker_connection_manager/init.pyhelpers/tests/下提供了test_observability_telemetry.py与test_observability_http_instrumentation.py两组测试sdk/packages/python/iii/tests/下还有test_stream_models.py、test_stream_types.py等测试可作为理解各类型行为边界的参考。HTTP 辅助模块iii_helpers.httpHTTP 模块解决的是引擎把原始请求数据交给 Python 处理器的适配问题函数内部无需手工解析 dict而是拿到带类型的请求/响应对象。http()把流式 handler 包装成引擎可直接调用的函数http接收一个回调(req, res) - HttpResponse | None返回一个 III 引擎可直接调用的异步函数。包装器负责把引擎下发的原始 dict或InternalHttpRequest转换成回调期望的强类型StreamRequest/StreamResponse对。签名http(callback: Callable[Awaitable[HttpResponse[Any] | None]])从源码看http的wrapper会依次处理三种输入形态见 http/init.py输入是InternalHttpRequest实例直接使用输入是dict从path_params、query_params、body、headers、method、response、request_body键还原出InternalHttpRequest其他类型原样透传。随后构造StreamResponse(internal.response)与StreamRequest(...)并调用用户回调。这意味着你只需关注类型化的HttpRequest视图底层的 WebSocket 响应通道由包装器代为维护。用法示例from iii_helpers.http import http, HttpResponse app.function(http_handler) http async def handler(req, res): # req: StreamRequest含 method / headers / path_params / query_params / body return HttpResponse(status_code200, body{ok: True})HttpInvocationConfigHTTP 外部函数调用配置该配置用于描述对外部 HTTP 服务如 Lambda、Cloudflare Workers的调用参数字段类型必填说明urlstr否调用的目标 URLmethodHttpMethod否HTTP 方法默认POSTtimeout_msint \| None否请求超时毫秒headersdict[str, str] \| None否附加请求头authHttpAuthConfig \| None否认证配置bearer / HMAC / API key其中HttpMethod是源码中定义的Literal[GET, POST, PUT, PATCH, DELETE]注意它与核心builtin_triggers的 HTTP 方法枚举不同后者还覆盖 HEAD/OPTIONS见 http/init.py。三种认证类型认证字段的值都指向环境变量名而不是明文密钥本身敏感信息不会出现在代码或配置里类型字段说明HttpAuthApiKeyheaderAPI key 的自定义请求头名value_key存放 API key 值的环境变量名API key 通过自定义请求头发送HttpAuthBearertoken_key存放 bearer token 的环境变量名Bearer token 认证HttpAuthHmacsecret_key存放 HMAC 共享密钥的环境变量名基于共享密钥的 HMAC 签名校验三个模型都有type判别字段Literal[api_key]/bearer/hmac并由HttpAuthConfig HttpAuthHmac | HttpAuthBearer | HttpAuthApiKey组成联合类型。HttpRequest与HttpResponseHttpRequest表示一个已缓冲的 HTTP 请求字段类型说明methodstr请求方法如GET、POSTpath_paramsdict[str, str]从匹配路由提取的路径参数query_paramsdict[str, str \| list[str]]URL 查询串参数headersdict[str, str \| list[str]]请求头bodyAny \| None已解析的请求体HttpResponse表示要返回的缓冲响应字段类型说明status_codeintHTTP 状态码headersdict[str, str]响应头bodyAny \| None响应体model_configAnyPydantic 模型配置值得注意的实现细节HttpResponse的 Pydantic 配置为ConfigDict(populate_by_nameTrue, arbitrary_types_allowedTrue)且status_code在序列化时使用别名statusCode与引擎侧和 Rust SDK 的线缆格式保持一致见 http/init.py。可观测性模块iii_helpers.observability该模块是 Python Worker 接入可观测性的核心涵盖结构化日志、OpenTelemetry 初始化与 span 操作。Logger以 OTel LogRecord 为出口的结构化日志器Logger把每条日志作为 OpenTelemetry LogRecord 发出每次调用都会自动捕获当前 trace 与 span 上下文让日志与分布式链路天然关联无需手工拼接 trace id。当 OTel 未初始化时Logger会优雅地回退到 Python 标准logging。源码实现见 observability/logger.py展示了两个关键设计四级方法info/warn/error/debug映射到 OTelSeverityNumberINFO9、WARN13、ERROR17、DEBUG5日志属性中自动携带service.name结构化数据以log.data属性整体附加trace_id / span_id 优先取显式构造参数否则取当前 span 上下文。推荐用法把结构化数据作为第二个参数传入from iii import Logger logger Logger() # 基础日志trace 上下文自动注入 logger.info(Worker connected) # 结构化上下文便于在可观测后端过滤、聚合、建仪表盘 logger.info(Order processed, {order_id: ord_123, amount: 49.99, currency: USD}) logger.warn(Retry attempt, {attempt: 3, max_retries: 5, endpoint: /api/charge}) logger.error(Payment failed, { order_id: ord_123, gateway: stripe, error_code: card_declined, })官方文档特别强调使用 dict 键值对而不是字符串插值才能在 Grafana、Datadog 等后端做过滤与聚合。另外Logger构造函数还接受可选的trace_id/span_id/service_name用于在缺少活动 span 时显式绑定上下文。OtelConfigOpenTelemetry 初始化配置init_otel(config: OtelConfig | None None, loop: None None)负责初始化 OpenTelemetry后续重复调用为 no-op。其配置字段源码见 observability/telemetry_types.py字段类型默认值 / 说明enabledbool \| None是否启用 OTel默认 True设OTEL_ENABLEDfalse/0/no/off可关闭service_namestr \| None服务名默认取环境变量OTEL_SERVICE_NAME否则iii-python-sdkservice_versionstr \| None服务版本默认取SERVICE_VERSION否则unknownservice_namespacestr \| None服务命名空间属性service_instance_idstr \| None服务实例 ID默认随机 UUIDengine_ws_urlstr \| NoneIII 引擎 WebSocket 地址默认取环境变量III_URL否则ws://localhost:49134fetch_instrumentation_enabledbool是否通过URLLibInstrumentor自动埋点 urllib HTTP 调用默认 Truespans_flush_interval_msint \| Nonespan 处理器刷盘延迟默认 100ms环境变量覆盖OTEL_SPANS_FLUSH_INTERVAL_MS。文档特别提示OpenTelemetry 默认的 5000ms 正是操作结束后几秒才看到 trace的原因logs_enabledbool \| None是否经EngineLogExporter导出 OTel 日志OTel 启用时默认 Truelogs_flush_interval_msint \| None日志处理器刷盘延迟默认 100mslogs_batch_sizeint \| None每批导出的日志条数上限默认 1metrics_enabledbool是否经EngineMetricsExporter导出指标默认 Truemetrics_export_interval_msint指标导出间隔毫秒默认 6000060 秒ReconnectionConfigWebSocket 重连策略字段类型默认值说明initial_delay_msint1000起始延迟毫秒max_delay_msint30000最大延迟上限毫秒backoff_multiplierfloat2.0指数退避乘数jitter_factorfloat0.3随机抖动因子0–1max_retriesint-1最大重试次数-1表示无限重试Span 工具函数函数签名要点行为current_trace_id()无参返回当前活动 trace_id 的 32 位十六进制字符串不可用时返回 Nonecurrent_span_id()无参返回当前活动 span_id 的 16 位十六进制字符串不可用时返回 Nonecurrent_span_is_recording()无参无活动 span 或采样器丢弃 span 时返回 Falsewith_span(name, fn, kindNone, traceparentNone)异步启动新 span 并在其中运行fn(span)tracer 未初始化时用 no-op span 调用fn静默忽略属性/事件调用set_current_span_attribute(key, value)同步给当前 span 设置属性span 未 recording 时为 no-oprecord_span_event(name, attrsNone)同步记录 span 事件span 未 recording 时为 no-opset_current_span_error(message)同步标记当前 span 错误无活动 span 时为 no-op追踪上下文注入与提取跨服务传播追踪信息时使用 W3C 标准头函数行为inject_traceparent()把当前 trace 上下文注入 W3Ctraceparent头字符串extract_traceparent(traceparent: str)从 W3Ctraceparent头字符串提取 trace 上下文inject_baggage()把当前 baggage 注入 W3Cbaggage头字符串extract_baggage(baggage: str)从 W3Cbaggage头字符串提取 baggage配套的BaggageSpanProcessor负责把 baggage 中的键值同步到 span 上保证跨服务调用时业务上下文不丢失。生命周期与带追踪的 HTTP 调用init_otel(configNone, loopNone)初始化 OpenTelemetry重复调用为 no-opshutdown_otel()同步关闭 OTel尽力而为不等待 WS 刷盘flush_otel()shutdown_otel的对立面——在不拆除提供者的前提下强制刷盘所有 OTel 提供者。适合在短生命周期进程退出前使用既希望挂起的 spans/metrics/logs 被送达又计划继续使用 OTelexecute_traced_request(client, request)在 OTel CLIENT span 内执行 httpx 请求。具体行为包括向出站请求头注入 W3Ctraceparent在 span 上记录 HTTP 语义约定属性对 status 400 的响应设置 ERROR span 状态对网络级错误记录异常。日志脱敏与截断可观测性数据经常包含敏感字段模块提供了脱敏工具函数说明redact(value: Any)对值进行脱敏处理redact_and_truncate(value: Any, max_bytes: Optional[int] None)脱敏并按字节上限截断resolve_max_bytes_from_env()从环境变量解析截断上限DEFAULT_ALLOWLISTtuple[str, ...]类型的默认放行键集合这些键不被脱敏队列模块iii_helpers.queueEnqueueResult是函数以TriggerAction.Enqueue方式被调用时返回的结果类型字段类型说明messageReceiptIdstr引擎为入队任务分配的 UUID 回执 IDfrom iii_helpers.queue import EnqueueResult # 在函数内把消息投递到队列后返回回执 return EnqueueResult(messageReceiptIdreceipt_id)Stream 模块iii_helpers.streamStream 是 III 中流式数据 状态的核心抽象该模块提供四类类型触发器配置、变更事件、IO 输入输出、更新操作。Stream 触发器配置StreamTriggerConfig用于配置stream触发器决定哪些条目变更会触发 handler字段类型必填说明stream_namestr是要监听的流名称只有该流上的变更触发 handlergroup_idstr \| None否设置后仅该组内的变更触发item_idstr \| None否设置后仅该条目的变更触发condition_function_idstr \| None否条件函数 ID返回 False 时跳过 handlerStreamJoinLeaveTriggerConfig用于stream:join/stream:leave触发器仅含condition_function_id一个可选字段。Stream 事件类型StreamChangeEvent是stream触发器的 handler 输入在条目通过stream::set/stream::update/stream::delete变更时触发字段类型必填说明typeLiteral[stream]是事件类型timestampint是事件的 Unix 时间戳streamNamestr是变更发生的流groupIdstr是变更发生的组idstr \| None否变更的条目 IDeventStreamChangeEventDetail是变更详情StreamChangeEventDetail包含type: Literal[create, update, delete]变更类型与data: Any关联数据。StreamJoinLeaveEvent表示流的加入/离开事件字段包括subscription_id唯一订阅标识、stream_name、group_id、id可选条目 ID与context来自StreamAuthResult的认证上下文。认证相关类型类型字段说明StreamAuthInputheaders: dict[str, str]、path: str、query_params: dict[str, list[str]]、addr: str均必填流认证输入StreamAuthResultcontext: Any \| None认证通过后传给 stream handler 的任意上下文StreamJoinResultunauthorized: bool必填加入是否未授权Stream IO 输入输出操作输入字段结果字段Getstream_name、group_id、item_id—Setstream_name、group_id、item_id、data: Anyold_value: TData \| None、new_value: TDataDeletestream_name、group_id、item_idold_value: Any \| NoneListstream_name、group_id—ListGroupsstream_name—Updatestream_name、group_id、item_id、ops: list[UpdateOp]原子应用的有序操作列表old_value、new_value、errors: list[UpdateOpError]更新操作Update OpsUpdateOp是六种操作的联合类型UpdateSet | UpdateIncrement | UpdateDecrement | UpdateAppend | UpdateRemove | UpdateMerge。操作类型标识字段语义UpdateSetsetpath: str、value: Any把路径字段设为指定值空字符串表示根值UpdateIncrementincrementpath: str、by: int \| float数值字段增加byUpdateDecrementdecrementpath: str、by: int \| float数值字段减少byUpdateRemoveremovepath: str删除路径字段UpdateAppendappendpath: MergePath \| None、value: Any数组追加 / 字符串拼接 / 嵌套路径 pushUpdateMergemergepath: MergePath \| None、value: Any把对象浅合并到目标节点MergePath str | list[str]即单字符串旧式/一级键或字面量段列表嵌套路径与 Node SDK 的string | string[]别名及 Rust 的MergePath枚举对齐。UpdateAppend的路径形式与引擎语义接受的路径形式与UpdateMerge一致None//[]在根节点追加foo在一级键foo处追加。注意带点的字符串如a.b是字面量键名a.b不会被拆解成a - b[a, b, c]嵌套路径每个元素都是字面量段。引擎在叶子节点的行为嵌套路径上的缺失/非对象中间节点自动以{}创建/替换叶子缺失/null 嵌套路径 →[value]总是数组叶子缺失/null 单字符串路径 → 字符串拼接层级按字符串处理否则[value]已存在数组 → push已存在字符串 字符串值 → 拼接叶子是对象/标量 → 返回append.type_mismatch。UpdateMerge的路径形式与引擎语义根合并None//[]foo等价于[foo]即一级键[a, b, c]嵌套路径每个元素是字面量键[a.b]写入的是名为a.b的单个键而不是a - b。引擎行为路径上的缺失/非对象中间节点自动替换为{}合并是目标节点处的浅合并value的顶层键覆盖同名键兄弟键保留。安全与合法性校验重点merge与append都会做严格的输入校验违规操作返回结构化错误且该操作不会生效路径深度 32 段段长度 256 字节值深度 16merge顶层键 1024 个merge任何__proto__/constructor/prototype段或顶层键防原型污染。错误通过state::update/stream::update响应的errors数组返回成功应用的操作仍反映在响应的new_value中。UpdateOpError每个失败操作的错误详情字段类型说明op_indexint出错操作在原始ops数组中的下标codestr稳定错误码如merge.path.too_deepmessagestr可读的错误描述含具体数值doc_urlstr \| None可选的文档链接实现细节UpdateAppend与UpdateMerge都通过model_serializer(modewrap)在序列化时省略path: None从而让线缆载荷与 Rust SDK 的#[serde(skip_serializing_if Option::is_none)]逐字节一致见 stream/init.py。StreamUpdateResult.errors为空时也会从 JSON 线缆中省略该字段。Worker 连接管理模块RBAC该模块为通过 RBAC 端口连接的 Worker 提供认证输入/输出与三类注册钩子的类型定义是构建多租户、细粒度权限控制的基础。AuthInput与AuthResultAuthInput是 WebSocket 升级时传给 RBAC 认证函数的输入包含升级请求的 HTTP 头、查询参数与客户端 IP字段类型说明headersdict[str, str]WebSocket 升级请求的 HTTP 头query_paramsdict[str, list[str]]升级 URL 的查询参数每个键对应值列表以支持重复键ip_addressstr连接客户端 IPAuthResult控制认证后的 Worker 可以调用哪些函数、注册哪些触发器以及向中间件转发什么上下文字段类型说明allowed_functionslist[str]在expose_functions之外额外允许的函数 IDforbidden_functionslist[str]即使匹配expose_functions也拒绝的函数 IDallowed_trigger_typeslist[str] \| None允许注册触发器的触发器类型 IDNone表示全部允许allow_trigger_type_registrationbool是否允许注册新的触发器类型allow_function_registrationbool是否允许注册新函数function_registration_prefixstr \| None应用于该 Worker 注册的所有函数 ID 的前缀contextdict[str, Any]每次调用转发给中间件的任意上下文从源码看见 worker_connection_manager/init.pyAuthResult还支持namespaces: dict[str, list[str]]字段用于按命名空间授予作用域化权限例如{orders: [svc::*]}值可以是精确函数 ID也可以是通配符支持裸写或match(...)写法。当namespaces为空时会话可声明任意命名空间仅受allowed_functions与expose_functions约束。注册钩子三类钩子分别在 Worker 通过 RBAC 端口注册函数、触发器、触发器类型时被调用返回映射后的结果或抛异常拒绝注册。省略的字段保留注册请求中的原始值。函数注册on_function_registration_function_idOnFunctionRegistrationInputfunction_id、description、metadata、context会话的认证上下文OnFunctionRegistrationResultfunction_id、description、metadata均为可选省略即保留原值。触发器注册on_trigger_registration_function_idOnTriggerRegistrationInputtrigger_id、trigger_type、function_id、config触发器专属配置、metadata、contextOnTriggerRegistrationResulttrigger_id、trigger_type、function_id、config均为可选映射结果。触发器类型注册on_trigger_type_registration_function_idOnTriggerTypeRegistrationInputtrigger_type_id、description、contextOnTriggerTypeRegistrationResulttrigger_type_id、description均为可选映射结果。从源码还可看到函数与触发器注册输入都包含namespace字段显式注册值缺省为default由于同一 ID 可以存在于多个命名空间钩子需要依据命名空间逐项授权。小结三类落地场景综合以上五个模块iii-helpers在 III 项目中的典型组合方式是HTTP 服务用http()包装 handler 获得强类型请求/响应配合HttpInvocationConfig与三种HttpAuth*配置调用外部 HTTP 服务密钥全部走环境变量可观测性用OtelConfig初始化 OTelLogger输出与 trace 关联的结构化日志with_span/set_current_span_attribute/execute_traced_request串联分布式调用flush_otel在短进程退出前保证数据送达状态与实时流用StreamTriggerConfig订阅变更通过StreamSetInput/StreamUpdateInput内含merge、append等原子操作与原型污染防护读写流数据并用worker_connection_manager的类型构建 RBAC 认证与注册策略把多租户权限落到函数、触发器与命名空间三个粒度上。如需继续深入可阅读同版本配套文档 docs/0-20-0/sdk-reference/ 下的 SDK 指南以及sdk/packages/python/iii/tests/下的test_stream_models.py、test_stream_types.py、test_rbac_workers.py等测试了解各类型在真实调用链中的行为。【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考