vercel-workers 版本演进全解析Python 侧 Vercel Queues 与 Worker Services 的 SDK 能力图谱【免费下载链接】vercelDevelop. Preview. Ship.项目地址: https://gitcode.com/gh_mirrors/ve/vercel导读vercel-workers是 Vercel 官方开源仓库中面向 Python 的队列与 Worker 服务 SDK提供send()消息发布、subscribe消费原语以及 Celery、Dramatiq、Django tasks 三类任务框架适配器。本文以仓库内 python/vercel-workers/CHANGELOG.md 为骨架逐版本梳理 0.0.10 → 0.0.25 的功能演进脉络并结合源码与示例验证每项变更背后的真实实现帮助读者快速判断该 SDK 当前具备哪些能力、底层如何工作以及如何在自己的 Python 服务中正确使用。版本脉络一个从初始导入到企业级队列 SDK的演进史vercel-workers的 CHANGELOG 记录了两个阶段0.0.10 的 Initial Release与随后持续迭代的 0.0.11 ~ 0.0.25每个版本均以patch形式推进。当前包版本为 0.0.25声明于 pyproject.toml 中要求requires-python 3.12。0.0.10Initial Release将vercel-workersPython 包初始导入 monorepo提供Celery、Django tasks、Dramatiq三套适配器用于对接 Vercel Queues。这一定位延续至今在 README.md 中有明确描述Python SDK for Vercel Queues and Vercel Worker Services核心原语即send()与subscribe。核心 API 层演进从client到_queue/的内部重构0.0.20 / 0.0.21框架逻辑下沉与队列 SDK 拆分0.0.20将 framework-specific 逻辑重构进vercel-workers0.0.21把 Python 队列 SDK 重构进_queue/目录。从当前源码结构看该重构的结果是 src/vercel/workers/_queue/ 成为队列能力的底层实现区包含文件职责client.py发送消息的同步/异步客户端send、send_asyncHTTP 调用 Vercel Queue Service V3 APIsubscribe.pysubscribe装饰器、Subscription注册表、Ack/RetryAfter指令、payload 类型校验callback.py回调解析CloudEvent v1beta 与 v2beta 两种格式的解析与识别receive.py消息接收、可见性超时处理send.py发送请求构建URL、鉴权头、幂等键、保留/延迟头types.pyWorkerJSONEncoder、MessageMetadata、SendMessageResult等类型定义exceptions.py全套错误类型而面向用户的公共入口 src/vercel/workers/init.py 重新导出QueueClient、AsyncQueueClient、send、subscribe、Ack、RetryAfter、WorkerJSONEncoder、get_wsgi_app、get_asgi_app等。公共 API 保持稳定内部实现被模块化——这是该 SDK 演进中最重要的架构决策。0.0.22新增 QueueClient 与 AsyncQueueClientQueueClient与AsyncQueueClient在 src/vercel/workers/_queue/client.py 中实现为需要以客户端对象方式管理队列连接的用户提供面向对象接口与函数式send()/send_async()并行。0.0.18从公共 API 移除consumerconsumer不再作为公开 API 暴露。当前公开导出列表见init.py 的__all__中确实已无consumer用户配置消费组改为在vercel.json的 worker service 声明中完成。send() 发送链路参数、请求头与底层实现0.0.21retention/delay 支持 timedelta0.0.21 用retention与delay取代了retention_seconds与delay_seconds并支持datetime.timedelta传参例如retentiontimedelta(hours6)。在 src/vercel/workers/_queue/send.py 中可见完整处理逻辑_resolve_duration_alias()同时兼容新旧参数名但同一语义不能同时传两个参数否则抛TypeError_duration_to_seconds()校验必须为非负有限数值timedelta会被转换为秒最终通过Vqs-Retention-Seconds、Vqs-Delay-Seconds请求头发送。0.0.12send API 增加 headerssubscribe() 支持 topic filter0.0.12 为 send API 增加headers参数自定义请求头会与Authorization、Content-Type合并subscribe()支持 topic 过滤。当前subscribe支持三种形式见 src/vercel/workers/_queue/subscribe.pysubscribe # 不限定 topic def worker(message, metadata): ... subscribe(topicevents) # 精确匹配 topic def billing_worker(message, metadata): ... subscribe(topic(user-*, lambda t: t.startswith(user-))) # 自定义谓词过滤 def user_worker(message, metadata): ...0.0.19部署固定deployment pinning对齐 TypeScript SDK0.0.19 将队列部署固定行为与 TypeScript SDK 对齐区分三种状态自动固定默认通过VERCEL_DEPLOYMENT_ID环境变量将消息固定到当前部署显式部署 ID调用时传入deployment_iddpl_xxx显式取消固定传入deployment_idNone。resolve_deployment_id()send.py的实现要点开发模式下VERCEL_WORKERS_IN_PROCESS1或VERCEL_QUEUE_TOKENvc-dev-token永不发送部署 ID与vercel dev行为一致未显式传参且环境变量缺失时抛出RuntimeError提示可传显式deployment_id或deployment_idNone显式退出固定固定后通过Vqs-Deployment-Id请求头随消息发送。0.0.13支持非标准库类型的编码0.0.13 为 send 支持了常见非标准库类型作为参数编码。其实现是 src/vercel/workers/_queue/types.py 中的WorkerJSONEncoderclass WorkerJSONEncoder(json.JSONEncoder): def default(self, o): match o: case UUID(): return str(o) case datetime() | date(): return o.isoformat() case Decimal(): return float(o) case _: return super().default(o)即UUID→ 字符串、datetime/date→ ISO 格式、Decimal→ 浮点数用户也可通过send(json_encoder...)传入自定义编码器。消息消费回调协议从 v1beta 到 v2beta0.0.14 → 0.0.17从 Queues V3 API 迁移到 v2beta triggers0.0.14python workers 迁移至Queues V3 API对应get_queue_base_path()默认的/api/v3/topic端点0.0.17python workers 迁移至v2beta triggers 私有路由。0.0.23v2beta 元数据回调的兜底实现0.0.23 为 v2betametadata-only 回调增加receive_message_by_id兜底当回调仅携带消息元数据队列名、消费组、消息 ID 等头信息而不含完整 payload 时SDK 会按消息 ID 主动拉取完整消息后再分发给订阅者。该逻辑位于 src/vercel/workers/_queue/callback.py 与client.py的_handle_queue_callback()中通过Ce-Type: com.vercel.queue.v2beta头识别 v2beta 回调is_v2beta_callback从Ce-Vqsqueuename、Ce-Vqsconsumergroup、Ce-Vqsmessageid、Ce-Vqsreceipthandle等头解析队列上下文parse_v2beta_callback若回调中无 receipt/payload则调用resolve_v2beta_message以receive_message_by_id补全兼容 v1beta CloudEvent 结构parse_cloudeventtype com.vercel.queue.v1beta两种格式共用同一套分发管线。可见性超时与自动续期回调处理遵循 Node 侧ConsumerGroupOptions的默认值源码注释明确说明Mirror the Node defaultsVQS_VISIBILITY_TIMEOUT可见性超时默认30 秒VQS_VISIBILITY_REFRESH_INTERVAL自动续期间隔默认10 秒。处理期间通过VisibilityExtender后台任务持续刷新消息可见性防止任务执行中消息被重复投递。显式重试与确认指令Ack / RetryAfter0.0.19worker 可返回或抛出 RetryAfter / Ack0.0.19 为 Python worker 增加显式重试与确认指令。在 subscribe.py 中实现class Ack(Exception): ... class RetryAfter(Exception): def __init__(self, delay, reasonNone): # delay 支持 int 秒或 timedelta负值会被截断为 0invoke_subscriptions()的语义返回或抛出RetryAfter(delay)→ 消息按 delay 秒后重试通过change_visibility设置可见性返回或抛出Ack→ 立即确认删除消息返回其他任何值含None→ 视为处理成功并确认。examples/basic/worker.py 展示了实际用法返回None即确认消息需要重试则return RetryAfter(60)。0.0.25支持 per-actor Dramatiq 重试选项最新版本 0.0.25 让 Dramatiq worker 能遵循每个 actor 级别配置的重试选项而非仅使用全局默认。这补齐了 Dramatiq 适配器与上游 Dramatiq 生态actor 级max_retries、retry_when等的对接能力。0.0.15Dramatiq 中间件与序列化修复处理 Dramatiq 中间件middlewares修复 UUID/Decimal/datetime 的序列化问题即前文WorkerJSONEncoder的覆盖场景。认证与运行环境token 解析与进程内开发模式send()的 token 解析顺序send.py 的get_queue_token显式token参数VERCEL_QUEUE_TOKEN环境变量Vercel OIDC tokenvercel.oidc.get_vercel_oidc_token用于部署环境自动认证。队列端点解析get_queue_base_url优先VERCEL_QUEUE_BASE_URL其次若设置VERCEL_REGION则路由到https://{region}.vercel-queue.com如iad1兜底https://vercel-queue.com路径默认/api/v3/topic。进程内开发模式设置VERCEL_WORKERS_IN_PROCESS1时send()会在当前进程内直接调用匹配的subscribe处理器_send_in_process不访问队列服务模拟 TypeScript 侧的本地 dev 体验——无持久化、无可见性超时、无重试。若未注册任何订阅或没有匹配 topic 的订阅会抛出带可用 topic 列表的明确错误便于快速排查配置不匹配。安装与快速上手安装pip install vercel-workers按需安装适配器 extraspip install vercel-workers[celery] pip install vercel-workers[dramatiq] pip install vercel-workers[django]依赖见 pyproject.tomlhttpx0.27.0、anyio4.0.0、pydantic2.7.0、python-dotenv、vercel0.3.7。Worker Service 部署形态vercel.json{ projectSettings: { framework: services }, experimentalServices: { web: { framework: fastapi, entrypoint: main.py, routePrefix: / }, worker: { type: worker, entrypoint: worker.py, topic: default, consumer: default } } }worker.py需要暴露 worker 定义subscribe函数、Celeryapp或 Dramatiqbroker并导入任务模块以完成处理器注册。参考 examples/basic/vercel.json 的完整示例其中topic使用topics数组形式声明且与main.py中send()的队列名保持一一对应。完整示例FastAPI 生产者 subscribe Worker生产者侧 examples/basic/main.pyimport worker # noqa: F401 # 导入以注册 subscribe 处理器 from fastapi import FastAPI from pydantic import BaseModel from vercel.workers import send QUEUE_NAME default # 必须与 vercel.json 中 worker 的 topic 一致 app FastAPI() app.post(/enqueue) def enqueue_job(body: EnqueueRequest): result send(QUEUE_NAME, {message: body.message}) return {queued: True, messageId: result[messageId], queue: QUEUE_NAME}消费侧 examples/basic/worker.pyfrom vercel.workers import MessageMetadata, RetryAfter, subscribe subscribe(topicdefault) def process_message(message, metadata: MessageMetadata) - RetryAfter | None: print(Received message from queue:, message) return None # None 确认RetryAfter(60) 60 秒后重试错误处理体系exceptions.py 与公共导出提供了一整套结构化错误类型便于精确处理队列调用失败基础类VQSError含status_code、可选retry_after4xx 类BadRequestError400、UnauthorizedError401、ForbiddenError403、DuplicateIdempotencyKeyError409幂等键冲突、InvalidLimitError队列/消息状态类QueueEmptyError、MessageNotFoundError、MessageNotAvailableError、MessageCorruptedError、MessageLockedError其他InternalServerError5xx、ThrottledError限流、TokenResolutionErrortoken 无法解析。回调处理管线对VQSError会按其status_code原样回传 HTTP 状态码并附上type与可选retryAfter字段保证消费者侧错误语义一致。环境变量速查表环境变量作用默认值VERCEL_QUEUE_TOKEN队列鉴权 token部署外运行必需无VERCEL_QUEUE_BASE_URL队列服务端点覆盖https://vercel-queue.com或按 regionVERCEL_REGION按 region 路由队列端点无VERCEL_QUEUE_BASE_PATHAPI 路径前缀/api/v3/topicVERCEL_DEPLOYMENT_ID自动部署固定无VERCEL_WORKERS_IN_PROCESS启用进程内开发模式1/true/yes关闭VQS_VISIBILITY_TIMEOUT可见性超时秒30VQS_VISIBILITY_REFRESH_INTERVAL可见性续期间隔秒10在 Vercel 之外运行本地开发时至少需要设置VERCEL_QUEUE_TOKEN可选VERCEL_QUEUE_BASE_URL这是 README.md 明确说明的运行前提。测试与示例资产仓库提供了覆盖各能力面的测试与示例便于读者对照验证测试tests/test_client_and_callback.py、tests/test_celery_adapter.py、tests/test_dramatiq_adapter.py、tests/test_django_adapter.py、tests/test_runtime_bridge.py示例examples/basicFastAPI subscribe、examples/celery、examples/dramatiq、examples/django。小结从 0.0.10 到 0.0.25vercel-workers完成了三件关键事一是架构上把队列 SDK 下沉到_queue/并保持公共 API 稳定二是消息协议上从 v1beta CloudEvent 演进到 v2beta triggers并补齐 metadata-only 回调的按 ID 拉取兜底三是消费语义上引入Ack/RetryAfter显式指令与 deployment pinning使其行为与 TypeScript SDK 对齐。对于需要在 Vercel 上用 Python 构建异步任务系统的开发者该 SDK 当前已具备生产可用的发布、消费、重试、鉴权与多框架适配能力可直接参照上述配置与示例落地。【免费下载链接】vercelDevelop. Preview. Ship.项目地址: https://gitcode.com/gh_mirrors/ve/vercel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考