跳转至

tokens

feishu.auth.tokens

CachedToken dataclass

缓存中的访问凭证。

参数：

名称 类型 描述 默认

value

str

访问凭证。

 必需

expire_at

float

凭证的过期时刻，与 feishu.auth.tokens.TokenManager 使用的时钟同源。

 必需

示例：

Python Console Session
1
2
3
4
5
>>> token = CachedToken("t-1", 5400.0)
>>> token.value
't-1'
>>> token.expire_at
5400.0
源代码位于： feishu/auth/tokens.py
Python
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@dataclass
class CachedToken:
    r"""
    缓存中的访问凭证。

    Args:
        value: 访问凭证。
        expire_at: 凭证的过期时刻，与 [feishu.auth.tokens.TokenManager][] 使用的时钟同源。

    Examples:
        >>> token = CachedToken("t-1", 5400.0)
        >>> token.value
        't-1'
        >>> token.expire_at
        5400.0
    """

    value: str
    expire_at: float

TokenCache

Bases: Protocol

访问凭证缓存协议。

实现该协议即可作为 feishu.auth.tokens.TokenManager 的凭证缓存后端。 默认实现为进程内缓存 feishu.auth.tokens.InMemoryTokenCache；如需在多个进程或 实例间共享凭证，可自行实现基于 Redis 等外部存储的缓存。

源代码位于： feishu/auth/tokens.py
Python
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class TokenCache(Protocol):
    r"""
    访问凭证缓存协议。

    实现该协议即可作为 [feishu.auth.tokens.TokenManager][] 的凭证缓存后端。
    默认实现为进程内缓存 [feishu.auth.tokens.InMemoryTokenCache][]；如需在多个进程或
    实例间共享凭证，可自行实现基于 Redis 等外部存储的缓存。
    """

    async def get(self, key: str) -> CachedToken | None:
        r"""
        读取缓存的凭证。

        Args:
            key: 由 [feishu.auth.credentials.Credential.cache_key][] 生成的缓存键。

        Returns:
            命中的凭证；未命中时返回 `None`。
        """

    async def set(self, key: str, token: CachedToken) -> None:
        r"""
        写入缓存的凭证。

        Args:
            key: 由 [feishu.auth.credentials.Credential.cache_key][] 生成的缓存键。
            token: 待缓存的凭证。
        """

get async

Python
get(key: str) -> CachedToken | None

读取缓存的凭证。

参数：

名称 类型 描述 默认
key
str

由 feishu.auth.credentials.Credential.cache_key 生成的缓存键。

 必需

返回：

类型 描述
CachedToken | None

命中的凭证；未命中时返回 None。
源代码位于： feishu/auth/tokens.py
Python
67
68
69
70
71
72
73
74
75
76
async def get(self, key: str) -> CachedToken | None:
    r"""
    读取缓存的凭证。

    Args:
        key: 由 [feishu.auth.credentials.Credential.cache_key][] 生成的缓存键。

    Returns:
        命中的凭证；未命中时返回 `None`。
    """

set async

Python
set(key: str, token: CachedToken) -> None

写入缓存的凭证。

参数：

名称 类型 描述 默认
key
str

由 feishu.auth.credentials.Credential.cache_key 生成的缓存键。

 必需
token
CachedToken

待缓存的凭证。

 必需
源代码位于： feishu/auth/tokens.py
Python
78
79
80
81
82
83
84
85
async def set(self, key: str, token: CachedToken) -> None:
    r"""
    写入缓存的凭证。

    Args:
        key: 由 [feishu.auth.credentials.Credential.cache_key][] 生成的缓存键。
        token: 待缓存的凭证。
    """

InMemoryTokenCache

进程内访问凭证缓存。

feishu.auth.tokens.TokenCache 的默认实现，将凭证保存在进程内存中。 凭证不会在进程间共享，进程退出后即失效；如需跨进程共享，请自行实现 TokenCache 协议。

示例：

Python Console Session
1
2
3
4
5
6
7
>>> import asyncio
>>> cache = InMemoryTokenCache()
>>> asyncio.run(cache.set("k", CachedToken("t-1", 5400.0)))
>>> asyncio.run(cache.get("k")).value
't-1'
>>> asyncio.run(cache.get("missing")) is None
True
源代码位于： feishu/auth/tokens.py
Python
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class InMemoryTokenCache:
    r"""
    进程内访问凭证缓存。

    [feishu.auth.tokens.TokenCache][] 的默认实现，将凭证保存在进程内存中。
    凭证不会在进程间共享，进程退出后即失效；如需跨进程共享，请自行实现 `TokenCache` 协议。

    Examples:
        >>> import asyncio
        >>> cache = InMemoryTokenCache()
        >>> asyncio.run(cache.set("k", CachedToken("t-1", 5400.0)))
        >>> asyncio.run(cache.get("k")).value
        't-1'
        >>> asyncio.run(cache.get("missing")) is None
        True
    """

    def __init__(self) -> None:
        self._store: dict[str, CachedToken] = {}

    async def get(self, key: str) -> CachedToken | None:
        r"""
        读取缓存的凭证。

        Args:
            key: 缓存键。

        Returns:
            命中的凭证；未命中时返回 `None`。
        """
        return self._store.get(key)

    async def set(self, key: str, token: CachedToken) -> None:
        r"""
        写入缓存的凭证。

        Args:
            key: 缓存键。
            token: 待缓存的凭证。
        """
        self._store[key] = token

get async

Python
get(key: str) -> CachedToken | None

读取缓存的凭证。

参数：

名称 类型 描述 默认
key
str

缓存键。

 必需

返回：

类型 描述
CachedToken | None

命中的凭证；未命中时返回 None。
源代码位于： feishu/auth/tokens.py
Python
108
109
110
111
112
113
114
115
116
117
118
async def get(self, key: str) -> CachedToken | None:
    r"""
    读取缓存的凭证。

    Args:
        key: 缓存键。

    Returns:
        命中的凭证；未命中时返回 `None`。
    """
    return self._store.get(key)

set async

Python
set(key: str, token: CachedToken) -> None

写入缓存的凭证。

参数：

名称 类型 描述 默认
key
str

缓存键。

 必需
token
CachedToken

待缓存的凭证。

 必需
源代码位于： feishu/auth/tokens.py
Python
120
121
122
123
124
125
126
127
128
async def set(self, key: str, token: CachedToken) -> None:
    r"""
    写入缓存的凭证。

    Args:
        key: 缓存键。
        token: 待缓存的凭证。
    """
    self._store[key] = token

TokenManager

应用级访问凭证管理器。

在凭据与传输层之上提供凭证的缓存、过期前刷新与并发去重：同一凭证只会被换取一次并复用， 临近过期时自动重新换取，且一批并发调用只会触发一次换取（避免惊群）。 不同凭证类型（如 tenant 与 app）使用各自独立的锁，互不阻塞。

参数：

名称 类型 描述 默认

credential

Credential

用于换取凭证的应用凭据，参见 feishu.auth.credentials.Credential。

 必需

transport

Transport

用于发起请求的传输层。

 必需

cache

TokenCache | None

凭证缓存后端。默认为 feishu.auth.tokens.InMemoryTokenCache。

 None

refresh_offset

int

过期前的提前刷新秒数，凭证在距过期不足该秒数时即重新换取。

 TOKEN_REFRESH_OFFSET

min_ttl

int

扣除 refresh_offset 后的最小有效缓存时长（秒）。当凭证有效期较短 （expire <= refresh_offset）时，过期时刻本会落在当前时刻或之前，导致下一次读取 立刻重新换取（惊群）；此时改为缓存 min_ttl 秒，确保短期凭证仍被短暂复用。

 MIN_TOKEN_TTL

now

Callable[[], float]

返回当前时刻的单调时钟函数，主要用于测试注入。

 monotonic
飞书文档

tenant_access_token: https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token

app_access_token: https://open.feishu.cn/document/server-docs/authentication-management/access-token/app_access_token

源代码位于： feishu/auth/tokens.py
Python
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class TokenManager:
    r"""
    应用级访问凭证管理器。

    在凭据与传输层之上提供凭证的缓存、过期前刷新与并发去重：同一凭证只会被换取一次并复用，
    临近过期时自动重新换取，且一批并发调用只会触发一次换取（避免惊群）。
    不同凭证类型（如 `tenant` 与 `app`）使用各自独立的锁，互不阻塞。

    Args:
        credential: 用于换取凭证的应用凭据，参见 [feishu.auth.credentials.Credential][]。
        transport: 用于发起请求的传输层。
        cache: 凭证缓存后端。默认为 [feishu.auth.tokens.InMemoryTokenCache][]。
        refresh_offset: 过期前的提前刷新秒数，凭证在距过期不足该秒数时即重新换取。
        min_ttl: 扣除 `refresh_offset` 后的最小有效缓存时长（秒）。当凭证有效期较短
            （`expire <= refresh_offset`）时，过期时刻本会落在当前时刻或之前，导致下一次读取
            立刻重新换取（惊群）；此时改为缓存 `min_ttl` 秒，确保短期凭证仍被短暂复用。
        now: 返回当前时刻的单调时钟函数，主要用于测试注入。

    飞书文档:
        `tenant_access_token`:
        https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token

        `app_access_token`:
        https://open.feishu.cn/document/server-docs/authentication-management/access-token/app_access_token
    """

    def __init__(
        self,
        credential: Credential,
        transport: Transport,
        *,
        cache: TokenCache | None = None,
        refresh_offset: int = TOKEN_REFRESH_OFFSET,
        min_ttl: int = MIN_TOKEN_TTL,
        now: Callable[[], float] = time.monotonic,
    ) -> None:
        self._credential = credential
        self._transport = transport
        self._cache = cache or InMemoryTokenCache()
        self._refresh_offset = refresh_offset
        self._min_ttl = min_ttl
        self._now = now
        # One lock per cache key, so fetching one token type (e.g. tenant) never
        # blocks fetching another (e.g. a user token). Created lazily; safe to do
        # without a guard since asyncio is single-threaded and there is no await
        # between the check and the insert below.
        self._locks: dict[str, asyncio.Lock] = {}

    def _lock_for(self, key: str) -> asyncio.Lock:
        lock = self._locks.get(key)
        if lock is None:
            lock = self._locks[key] = asyncio.Lock()
        return lock

    async def tenant_access_token(self) -> str:
        r"""
        获取租户访问凭证（`tenant_access_token`）。

        等价于 `token("tenant")`，命中缓存且未临近过期时直接复用，否则重新换取。

        Returns:
            租户访问凭证。

        飞书文档:
            `tenant_access_token`:
            https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token

        Examples:
            >>> token = await manager.tenant_access_token()  # doctest: +SKIP
            >>> token  # doctest: +SKIP
            't-xxxxxxxx'
        """
        return await self.token("tenant")

    async def token(self, token_type: str = "tenant") -> str:
        r"""
        获取指定类型的应用级访问凭证。

        命中缓存且距过期不少于 `refresh_offset` 秒时直接复用缓存值，否则换取新凭证并写入缓存。
        同一凭证类型上的并发调用会在锁内做双重检查，最终只触发一次换取。

        Args:
            token_type: 凭证类型，`tenant` 或 `app`。默认为 `tenant`。

        Returns:
            对应类型的访问凭证。

        飞书文档:
            `tenant_access_token`:
            https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token

            `app_access_token`:
            https://open.feishu.cn/document/server-docs/authentication-management/access-token/app_access_token

        Examples:
            >>> token = await manager.token("app")  # doctest: +SKIP
            >>> token  # doctest: +SKIP
            'a-xxxxxxxx'
        """
        key = self._credential.cache_key(token_type, self._transport.base_url)
        cached = await self._cache.get(key)
        if cached is not None and cached.expire_at > self._now():
            return cached.value
        async with self._lock_for(key):
            cached = await self._cache.get(key)  # double-check after acquiring
            if cached is not None and cached.expire_at > self._now():
                return cached.value
            value, expire = await self._credential.fetch(self._transport, token_type)
            # Clamp the effective TTL: a short token (expire <= refresh_offset) would
            # otherwise yield expire_at <= now and be re-fetched on the next read.
            ttl = max(expire - self._refresh_offset, self._min_ttl)
            expire_at = self._now() + ttl
            await self._cache.set(key, CachedToken(value, expire_at))
            return value

tenant_access_token async

Python
tenant_access_token() -> str

获取租户访问凭证（tenant_access_token）。

等价于 token("tenant")，命中缓存且未临近过期时直接复用，否则重新换取。

返回：

类型 描述
str

租户访问凭证。
飞书文档

tenant_access_token: https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token

示例：

Python Console Session
1
2
3
>>> token = await manager.tenant_access_token()
>>> token
't-xxxxxxxx'
源代码位于： feishu/auth/tokens.py
Python
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
async def tenant_access_token(self) -> str:
    r"""
    获取租户访问凭证（`tenant_access_token`）。

    等价于 `token("tenant")`，命中缓存且未临近过期时直接复用，否则重新换取。

    Returns:
        租户访问凭证。

    飞书文档:
        `tenant_access_token`:
        https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token

    Examples:
        >>> token = await manager.tenant_access_token()  # doctest: +SKIP
        >>> token  # doctest: +SKIP
        't-xxxxxxxx'
    """
    return await self.token("tenant")

token async

Python
token(token_type: str = 'tenant') -> str

获取指定类型的应用级访问凭证。

命中缓存且距过期不少于 refresh_offset 秒时直接复用缓存值，否则换取新凭证并写入缓存。 同一凭证类型上的并发调用会在锁内做双重检查，最终只触发一次换取。

参数：

名称 类型 描述 默认
token_type
str

凭证类型，tenant 或 app。默认为 tenant。

 'tenant'

返回：

类型 描述
str

对应类型的访问凭证。
飞书文档

tenant_access_token: https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token

app_access_token: https://open.feishu.cn/document/server-docs/authentication-management/access-token/app_access_token

示例：

Python Console Session
1
2
3
>>> token = await manager.token("app")
>>> token
'a-xxxxxxxx'
源代码位于： feishu/auth/tokens.py
Python
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
async def token(self, token_type: str = "tenant") -> str:
    r"""
    获取指定类型的应用级访问凭证。

    命中缓存且距过期不少于 `refresh_offset` 秒时直接复用缓存值，否则换取新凭证并写入缓存。
    同一凭证类型上的并发调用会在锁内做双重检查，最终只触发一次换取。

    Args:
        token_type: 凭证类型，`tenant` 或 `app`。默认为 `tenant`。

    Returns:
        对应类型的访问凭证。

    飞书文档:
        `tenant_access_token`:
        https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token

        `app_access_token`:
        https://open.feishu.cn/document/server-docs/authentication-management/access-token/app_access_token

    Examples:
        >>> token = await manager.token("app")  # doctest: +SKIP
        >>> token  # doctest: +SKIP
        'a-xxxxxxxx'
    """
    key = self._credential.cache_key(token_type, self._transport.base_url)
    cached = await self._cache.get(key)
    if cached is not None and cached.expire_at > self._now():
        return cached.value
    async with self._lock_for(key):
        cached = await self._cache.get(key)  # double-check after acquiring
        if cached is not None and cached.expire_at > self._now():
            return cached.value
        value, expire = await self._credential.fetch(self._transport, token_type)
        # Clamp the effective TTL: a short token (expire <= refresh_offset) would
        # otherwise yield expire_at <= now and be re-fetched on the next read.
        ttl = max(expire - self._refresh_offset, self._min_ttl)
        expire_at = self._now() + ttl
        await self._cache.set(key, CachedToken(value, expire_at))
        return value