credential-pool-sdk

一个基于 Redis 的高并发共享凭证池 SDK。

它的目标不是绑定某个具体业务，而是提供一套可复用的“凭证池基础设施”能力，适合在多进程、多节点环境下统一管理以下对象：

1. Cookie
2. JWT
3. Token
4. Nonce
5. Session
6. 其他需要共享、复用、并发控制的认证凭证

设计目标

本项目聚焦“池能力”，不耦合业务代码，不内置具体站点逻辑。

核心目标：

1. 通过 pip install 安装使用
2. 只依赖 Redis 作为共享状态中心
3. 提供统一的凭证池 API
4. 支持高并发随机获取
5. 支持独占获取与归还
6. 支持失败计数、租约回收、统计查询

适用场景

适合以下场景：

1. 多个 worker 共享一批 Cookie / JWT
2. 某些凭证允许被重复并发使用
3. 某些凭证同一时间只能被一个 worker 独占使用
4. 多台机器共同消费同一池凭证
5. 业务项目希望把“凭证共享与并发控制”抽成独立 SDK

架构图

下面这张图描述了 credential-pool-sdk 在业务项目中的典型位置：

flowchart LR
    A[业务 Producer<br/>生成 Cookie / JWT / Token] --> B[credential-pool-sdk]
    C[业务 Consumer 1<br/>Worker / Collector] --> B
    D[业务 Consumer 2<br/>Worker / Service] --> B
    E[业务 Consumer N<br/>多节点实例] --> B

    B --> F[(Redis)]

    subgraph SDK[credential-pool-sdk]
        B1[RedisCredentialPool]
        B2[CredentialItem]
        B3[Lease / Failure / Stats]
    end

    B --- B1
    B --- B2
    B --- B3

架构说明

1. Producer 负责生成凭证，并调用 SDK 写入池。
2. Consumer 负责从池中获取凭证，并在使用后回写结果。
3. SDK 只负责“共享池能力”，不负责业务调度。
4. Redis 作为多节点之间的共享状态中心。

时序图

下面是“独占获取一个当前未被使用的凭证，并在使用后归还”的典型时序：

sequenceDiagram
    participant Worker as 业务 Worker
    participant SDK as credential-pool-sdk
    participant Redis as Redis
    participant Target as 目标站点

    Worker->>SDK: acquire_available(owner, lease_ttl)
    SDK->>Redis: Lua 原子脚本<br/>从 available 随机取一个 key<br/>写入 lease_owner / lease_expiry
    Redis-->>SDK: 返回 credential key
    SDK->>Redis: HGET items[key]
    Redis-->>SDK: 返回 CredentialItem
    SDK-->>Worker: 返回独占凭证

    Worker->>Target: 使用凭证发起请求
    Target-->>Worker: 返回结果

    alt 成功
        Worker->>SDK: record_success(key)
        SDK->>Redis: HDEL failures[key]
        Worker->>SDK: release(key, owner)
        SDK->>Redis: Lua 原子脚本<br/>删除 lease<br/>放回 available
    else 失败
        Worker->>SDK: record_failure(key)
        SDK->>Redis: HINCRBY failures[key]
        Worker->>SDK: release(key, owner)
        SDK->>Redis: Lua 原子脚本<br/>删除 lease<br/>放回 available
    end

两种获取模式的区别

模式一：共享获取

item = await pool.random_get()

特点：

1. 同一个凭证可以被多个调用方同时拿到
2. 不会改变凭证当前可用状态
3. 适合可共享复用的凭证

模式二：独占获取

leased = await pool.acquire_available(owner="worker-1", lease_ttl=60)

特点：

1. 同一个凭证同一时间只能被一个调用方持有
2. 获取后必须通过 release() 归还
3. 若调用方崩溃，可通过 reclaim_expired_leases() 回收

核心能力

1. 统一对象模型

池中的元素不是裸字符串，而是统一对象，例如：

{
  "key": "jwt:1",
  "value": "token-1",
  "kind": "jwt",
  "status": "active",
  "created_at": 1760000000,
  "updated_at": 1760000000,
  "meta": {}
}

这样后续扩展字段时，不需要改动使用方式。

2. 两种获取模式

本 SDK 提供两种不同的“取凭证”方式。

random_get()

随机获取一个凭证。

特点：

1. 同一个凭证可以被多个调用方同时取到
2. 适合“可共享复用”的场景
3. 例如：某些 JWT、某些无状态 Cookie

acquire_available(owner, lease_ttl)

随机获取一个当前未被占用的凭证，并立即标记为“已被使用”。

特点：

1. 同一时间，一个凭证只能被一个调用方独占获取
2. 适合“同一时刻只能单独使用”的场景
3. 获取后必须通过 release() 归还
4. 若调用方异常退出，也可以通过过期租约回收

3. 归还凭证

通过：

await pool.release(key, owner="worker-1")

将独占中的凭证重新放回可用池。

4. 基础 CRUD

支持：

1. 添加单个凭证
2. 批量添加凭证
3. 查询单个凭证
4. 列表查询
5. 更新凭证
6. 删除凭证

5. 失败计数

支持：

1. record_success(key)
2. record_failure(key)
3. get_failure_count(key)

适合业务层自行实现失败阈值策略。

6. 租约回收

支持：

await pool.reclaim_expired_leases()

用于回收超时未归还的独占凭证。

7. 统计能力

支持：

1. 总凭证数
2. 可用凭证数
3. 已租约凭证数
4. 失败计数条目数

安装

pip install credential-pool-sdk

如果是本地开发：

pip install -e .

快速开始

创建连接

from credential_pool_sdk import RedisCredentialPool

pool = RedisCredentialPool.from_url(
    "redis://127.0.0.1:6379/0",
    pool_name="vimeo_jwt",
)

凭证对象示例

from credential_pool_sdk import CredentialItem

item = CredentialItem(
    key="jwt:1",
    value="token-1",
    kind="jwt",
    meta={"source": "producer-a"},
)

添加凭证

from credential_pool_sdk import CredentialItem

await pool.add(
    CredentialItem(
        key="jwt:1",
        value="token-1",
        kind="jwt",
    )
)

随机获取一个可共享凭证

item = await pool.random_get()
if item is not None:
    print(item.key, item.value)

独占获取一个当前未使用凭证

leased = await pool.acquire_available(owner="worker-1", lease_ttl=60)
if leased is not None:
    print("acquired:", leased.key)

归还独占凭证

await pool.release(leased.key, owner="worker-1")

失败计数

await pool.record_failure("jwt:1")
failures = await pool.get_failure_count("jwt:1")
print(failures)

回收超时未归还租约

reclaimed = await pool.reclaim_expired_leases()
print("reclaimed:", reclaimed)

完整示例

import asyncio

from credential_pool_sdk import CredentialItem, RedisCredentialPool


async def main() -> None:
    pool = RedisCredentialPool.from_url(
        "redis://127.0.0.1:6379/0",
        pool_name="vimeo_jwt",
    )

    await pool.add(
        CredentialItem(
            key="jwt:1",
            value="token-1",
            kind="jwt",
        )
    )

    shared_item = await pool.random_get()
    print("shared:", shared_item)

    leased_item = await pool.acquire_available(owner="worker-1", lease_ttl=60)
    print("leased:", leased_item)

    if leased_item is not None:
        await pool.release(leased_item.key, owner="worker-1")

    stats = await pool.stats()
    print(stats)

    await pool.close()


asyncio.run(main())

API 概览

池对象

RedisCredentialPool

主要方法：

1. add(item)
2. add_many(items)
3. get(key)
4. list(limit=100)
5. update(key, patch)
6. remove(key)
7. random_get()
8. acquire_available(owner, lease_ttl=60, exclude_keys=None)
9. release(key, owner=None)
10. reclaim_expired_leases(now_ts=None)
11. record_success(key)
12. record_failure(key, amount=1)
13. get_failure_count(key)
14. count()
15. count_available()
16. count_leased()
17. stats()
18. close()

数据对象

CredentialItem

字段：

1. key
2. value
3. kind
4. status
5. created_at
6. updated_at
7. meta

统计对象

PoolStats

字段：

1. total
2. available
3. leased
4. failures

Redis 数据模型

对于池名 {pool_name}，当前版本使用以下 Redis Key：

1. credential_pool:{pool_name}:items Redis Hash，存储 key -> json
2. credential_pool:{pool_name}:active Redis Set，存储所有激活凭证 key
3. credential_pool:{pool_name}:available Redis Set，存储当前可独占获取的凭证 key
4. credential_pool:{pool_name}:failures Redis Hash，存储 key -> failure_count
5. credential_pool:{pool_name}:lease_owner Redis Hash，存储 key -> owner
6. credential_pool:{pool_name}:lease_expiry Redis Sorted Set，存储 key -> expiry_ts

数据结构关系

flowchart TD
    A[items<br/>Hash: key -> json] --> B[active<br/>Set: 所有激活 key]
    B --> C[available<br/>Set: 当前可独占获取 key]
    A --> D[failures<br/>Hash: key -> failure_count]
    A --> E[lease_owner<br/>Hash: key -> owner]
    A --> F[lease_expiry<br/>ZSet: key -> expiry_ts]

说明

1. items 是主数据存储。
2. active 表示凭证仍在池内。
3. available 表示该凭证当前未被独占持有。
4. lease_owner 与 lease_expiry 一起构成租约系统。
5. failures 用于业务层实现失败阈值控制。

推荐使用模式

适合使用 random_get()

1. JWT 可被多个请求共享使用
2. 无状态 Cookie
3. 对单个凭证并发使用没有强限制的场景

适合使用 acquire_available()

1. 单个凭证同一时间只能被一个 worker 使用
2. 代理认证信息
3. 有明显并发冲突风险的 Cookie / Token
4. 需要“使用后归还”的场景

最佳实践

1. key 设计成全局唯一且可读，例如 jwt:1、cookie:user_001
2. kind 用于标识凭证类型，例如 jwt、cookie、nonce
3. 独占获取后务必在 finally 中执行 release()
4. 对长时间运行任务，建议定期执行 reclaim_expired_leases()
5. 失败阈值删除策略建议由业务层自行控制，不要在 SDK 核心层写死

License

待补充。