跳转至

user_tokens

feishu.auth.user_tokens

用户态访问凭证的持久化与按需刷新：把「以用户身份执行」做成一个标准接缝。

feishu.auth.oauth.OAuthNamespace 已封装授权 URL、exchange_code、refresh、user_info 等线缆调用，但 不负责「持久化每个用户的 user_access_token / refresh_token 并在临近过期时自动刷新」。本模块补上这一层： feishu.auth.user_tokens.OAuthTokenStore 抽象按用户多别名（open_id/union_id/user_id）存取凭证， feishu.auth.user_tokens.UserTokenProvider 据此解析并刷新凭证、产出用户态客户端（client.as_user(token)）。 其 as_user(user) 正好契合 feishu.agent.context.ToolContext 所需的提供方接口，可直接传给 feishu.agent.loop.AgentEngine 的 user_tokens 参数。

TokenRecord dataclass

一个用户的凭证记录：用户态访问令牌、（轮换式）刷新令牌、各项过期时间与多别名键。

飞书的 refresh_token 为一次性轮换：每次刷新都会得到全新的刷新令牌，旧令牌随即失效，因此刷新后必须 整体覆盖保存本记录。

示例：

Python Console Session
1
2
3
>>> rec = TokenRecord(user_access_token="u-acc", expires_at=_now() + 7200)
>>> rec.is_expired(_now())
False
源代码位于： feishu/auth/user_tokens.py
Python
 86
 87
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@dataclass
class TokenRecord:
    r"""
    一个用户的凭证记录：用户态访问令牌、（轮换式）刷新令牌、各项过期时间与多别名键。

    飞书的 `refresh_token` 为一次性轮换：每次刷新都会得到全新的刷新令牌，旧令牌随即失效，因此刷新后必须
    整体覆盖保存本记录。

    Examples:
        >>> rec = TokenRecord(user_access_token="u-acc", expires_at=_now() + 7200)
        >>> rec.is_expired(_now())
        False
    """

    user_access_token: str
    refresh_token: str | None = None
    expires_at: int | None = None
    refresh_token_expires_at: int | None = None
    scope: str | None = None
    user_keys: tuple[str, ...] = ()

    def is_expiring(self, skew_seconds: int, now: int) -> bool:
        r"""访问令牌是否将在 `skew_seconds` 内过期。"""
        return self.expires_at is not None and self.expires_at - now <= skew_seconds

    def is_expired(self, now: int) -> bool:
        r"""访问令牌是否已过期。"""
        return self.expires_at is not None and self.expires_at <= now

    def to_dict(self) -> dict[str, Any]:
        r"""序列化为可 JSON 化字典（[feishu.auth.user_tokens.TokenRecord.from_dict][] 的逆操作）。"""
        data = {
            "user_access_token": self.user_access_token,
            "refresh_token": self.refresh_token,
            "expires_at": self.expires_at,
            "refresh_token_expires_at": self.refresh_token_expires_at,
            "scope": self.scope,
            "user_keys": list(self.user_keys),
        }
        return data

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> TokenRecord:
        r"""从 [feishu.auth.user_tokens.TokenRecord.to_dict][] 产出的映射还原记录。"""
        return cls(
            user_access_token=str(data.get("user_access_token") or ""),
            refresh_token=data.get("refresh_token"),
            expires_at=data.get("expires_at"),
            refresh_token_expires_at=data.get("refresh_token_expires_at"),
            scope=data.get("scope"),
            user_keys=tuple(data.get("user_keys") or ()),
        )

    @classmethod
    def from_token_data(cls, token_data: Mapping[str, Any], keys: tuple[str, ...], *, now: int) -> TokenRecord:
        r"""从 `exchange_code` / `refresh` 的响应构造记录（`expires_in` 等折算为绝对过期时间）。"""
        expires_in = _coerce_int(token_data.get("expires_in"))
        refresh_expires_in = _coerce_int(token_data.get("refresh_token_expires_in"))
        refresh_token = token_data.get("refresh_token")
        scope = token_data.get("scope")
        return cls(
            user_access_token=str(token_data.get("access_token") or ""),
            refresh_token=str(refresh_token) if refresh_token else None,
            expires_at=(now + expires_in) if expires_in is not None else None,
            refresh_token_expires_at=(now + refresh_expires_in) if refresh_expires_in is not None else None,
            scope=str(scope) if scope else None,
            user_keys=tuple(keys),
        )

is_expiring

Python
is_expiring(skew_seconds: int, now: int) -> bool

访问令牌是否将在 skew_seconds 内过期。

源代码位于： feishu/auth/user_tokens.py
Python
107
108
109
def is_expiring(self, skew_seconds: int, now: int) -> bool:
    r"""访问令牌是否将在 `skew_seconds` 内过期。"""
    return self.expires_at is not None and self.expires_at - now <= skew_seconds

is_expired

Python
is_expired(now: int) -> bool

访问令牌是否已过期。

源代码位于： feishu/auth/user_tokens.py
Python
111
112
113
def is_expired(self, now: int) -> bool:
    r"""访问令牌是否已过期。"""
    return self.expires_at is not None and self.expires_at <= now

to_dict

Python
to_dict() -> dict[str, Any]

序列化为可 JSON 化字典（feishu.auth.user_tokens.TokenRecord.from_dict 的逆操作）。

源代码位于： feishu/auth/user_tokens.py
Python
115
116
117
118
119
120
121
122
123
124
125
def to_dict(self) -> dict[str, Any]:
    r"""序列化为可 JSON 化字典（[feishu.auth.user_tokens.TokenRecord.from_dict][] 的逆操作）。"""
    data = {
        "user_access_token": self.user_access_token,
        "refresh_token": self.refresh_token,
        "expires_at": self.expires_at,
        "refresh_token_expires_at": self.refresh_token_expires_at,
        "scope": self.scope,
        "user_keys": list(self.user_keys),
    }
    return data

from_dict classmethod

Python
from_dict(data: Mapping[str, Any]) -> TokenRecord

从 feishu.auth.user_tokens.TokenRecord.to_dict 产出的映射还原记录。

源代码位于： feishu/auth/user_tokens.py
Python
127
128
129
130
131
132
133
134
135
136
137
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> TokenRecord:
    r"""从 [feishu.auth.user_tokens.TokenRecord.to_dict][] 产出的映射还原记录。"""
    return cls(
        user_access_token=str(data.get("user_access_token") or ""),
        refresh_token=data.get("refresh_token"),
        expires_at=data.get("expires_at"),
        refresh_token_expires_at=data.get("refresh_token_expires_at"),
        scope=data.get("scope"),
        user_keys=tuple(data.get("user_keys") or ()),
    )

from_token_data classmethod

Python
from_token_data(token_data: Mapping[str, Any], keys: tuple[str, ...], *, now: int) -> TokenRecord

从 exchange_code / refresh 的响应构造记录（expires_in 等折算为绝对过期时间）。

源代码位于： feishu/auth/user_tokens.py
Python
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@classmethod
def from_token_data(cls, token_data: Mapping[str, Any], keys: tuple[str, ...], *, now: int) -> TokenRecord:
    r"""从 `exchange_code` / `refresh` 的响应构造记录（`expires_in` 等折算为绝对过期时间）。"""
    expires_in = _coerce_int(token_data.get("expires_in"))
    refresh_expires_in = _coerce_int(token_data.get("refresh_token_expires_in"))
    refresh_token = token_data.get("refresh_token")
    scope = token_data.get("scope")
    return cls(
        user_access_token=str(token_data.get("access_token") or ""),
        refresh_token=str(refresh_token) if refresh_token else None,
        expires_at=(now + expires_in) if expires_in is not None else None,
        refresh_token_expires_at=(now + refresh_expires_in) if refresh_expires_in is not None else None,
        scope=str(scope) if scope else None,
        user_keys=tuple(keys),
    )

OAuthTokenStore

Bases: Protocol

用户态凭证存储协议：按用户多别名存取 feishu.auth.user_tokens.TokenRecord。

内置实现为 feishu.auth.user_tokens.InMemoryOAuthTokenStore 与 feishu.auth.user_tokens.SqliteOAuthTokenStore。该协议标注了 runtime_checkable。

源代码位于： feishu/auth/user_tokens.py
Python
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
@runtime_checkable
class OAuthTokenStore(Protocol):
    r"""
    用户态凭证存储协议：按用户多别名存取 [feishu.auth.user_tokens.TokenRecord][]。

    内置实现为 [feishu.auth.user_tokens.InMemoryOAuthTokenStore][] 与
    [feishu.auth.user_tokens.SqliteOAuthTokenStore][]。该协议标注了 `runtime_checkable`。
    """

    async def get(self, user: Mapping[str, Any]) -> TokenRecord | None:
        r"""按用户的任一别名键读取凭证记录，未命中返回 `None`。"""
        ...

    async def save(
        self,
        token_data: Mapping[str, Any],
        *,
        user_info: Mapping[str, Any] | None = None,
        user_keys: tuple[str, ...] = (),
    ) -> tuple[str, ...]:
        r"""
        保存一次凭证响应；用户身份取自 `user_info` 或显式 `user_keys`，返回写入的别名键。

        Raises:
            ValueError: 当 `user_info` 与 `user_keys` 均无法提供任何用户身份
                （open_id/union_id/user_id）时抛出。
        """
        ...

get async

Python
get(user: Mapping[str, Any]) -> TokenRecord | None

按用户的任一别名键读取凭证记录，未命中返回 None。

源代码位于： feishu/auth/user_tokens.py
Python
165
166
167
async def get(self, user: Mapping[str, Any]) -> TokenRecord | None:
    r"""按用户的任一别名键读取凭证记录，未命中返回 `None`。"""
    ...

save async

Python
save(token_data: Mapping[str, Any], *, user_info: Mapping[str, Any] | None = None, user_keys: tuple[str, ...] = ()) -> tuple[str, ...]

保存一次凭证响应；用户身份取自 user_info 或显式 user_keys，返回写入的别名键。

引发：

类型 描述
ValueError

当 user_info 与 user_keys 均无法提供任何用户身份 （open_id/union_id/user_id）时抛出。
源代码位于： feishu/auth/user_tokens.py
Python
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def save(
    self,
    token_data: Mapping[str, Any],
    *,
    user_info: Mapping[str, Any] | None = None,
    user_keys: tuple[str, ...] = (),
) -> tuple[str, ...]:
    r"""
    保存一次凭证响应；用户身份取自 `user_info` 或显式 `user_keys`，返回写入的别名键。

    Raises:
        ValueError: 当 `user_info` 与 `user_keys` 均无法提供任何用户身份
            （open_id/union_id/user_id）时抛出。
    """
    ...

InMemoryOAuthTokenStore

基于内存的 feishu.auth.user_tokens.OAuthTokenStore 实现，仅适用于单进程、可接受重启即丢失的场景。

示例：

Python Console Session
1
2
>>> isinstance(InMemoryOAuthTokenStore(), OAuthTokenStore)
True
源代码位于： feishu/auth/user_tokens.py
Python
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
class InMemoryOAuthTokenStore:
    r"""
    基于内存的 [feishu.auth.user_tokens.OAuthTokenStore][] 实现，仅适用于单进程、可接受重启即丢失的场景。

    Examples:
        >>> isinstance(InMemoryOAuthTokenStore(), OAuthTokenStore)
        True
    """

    def __init__(self) -> None:
        self._by_key: dict[str, TokenRecord] = {}
        self._lock = asyncio.Lock()

    async def get(self, user: Mapping[str, Any]) -> TokenRecord | None:
        r"""参见 [feishu.auth.user_tokens.OAuthTokenStore.get][]。"""
        keys = user_identity_keys(user)
        async with self._lock:
            for key in keys:
                record = self._by_key.get(key)
                if record is not None:
                    return record
        return None

    async def save(
        self,
        token_data: Mapping[str, Any],
        *,
        user_info: Mapping[str, Any] | None = None,
        user_keys: tuple[str, ...] = (),
    ) -> tuple[str, ...]:
        r"""参见 [feishu.auth.user_tokens.OAuthTokenStore.save][]。"""
        keys = _keys_for_save(user_info, user_keys)
        record = TokenRecord.from_token_data(token_data, keys, now=_now())
        async with self._lock:
            # Reconcile aliases: drop prior rows for this identity so a re-auth with fewer aliases
            # cannot leave a stale alias pointing at the dead old record.
            stale: set[str] = set()
            for key in keys:
                old = self._by_key.get(key)
                if old is not None:
                    stale.update(old.user_keys)
            for alias in stale - set(keys):
                self._by_key.pop(alias, None)
            for key in keys:
                self._by_key[key] = record
        return keys

get async

Python
get(user: Mapping[str, Any]) -> TokenRecord | None

参见 feishu.auth.user_tokens.OAuthTokenStore.get。

源代码位于： feishu/auth/user_tokens.py
Python
206
207
208
209
210
211
212
213
214
async def get(self, user: Mapping[str, Any]) -> TokenRecord | None:
    r"""参见 [feishu.auth.user_tokens.OAuthTokenStore.get][]。"""
    keys = user_identity_keys(user)
    async with self._lock:
        for key in keys:
            record = self._by_key.get(key)
            if record is not None:
                return record
    return None

save async

Python
save(token_data: Mapping[str, Any], *, user_info: Mapping[str, Any] | None = None, user_keys: tuple[str, ...] = ()) -> tuple[str, ...]

参见 feishu.auth.user_tokens.OAuthTokenStore.save。

源代码位于： feishu/auth/user_tokens.py
Python
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
async def save(
    self,
    token_data: Mapping[str, Any],
    *,
    user_info: Mapping[str, Any] | None = None,
    user_keys: tuple[str, ...] = (),
) -> tuple[str, ...]:
    r"""参见 [feishu.auth.user_tokens.OAuthTokenStore.save][]。"""
    keys = _keys_for_save(user_info, user_keys)
    record = TokenRecord.from_token_data(token_data, keys, now=_now())
    async with self._lock:
        # Reconcile aliases: drop prior rows for this identity so a re-auth with fewer aliases
        # cannot leave a stale alias pointing at the dead old record.
        stale: set[str] = set()
        for key in keys:
            old = self._by_key.get(key)
            if old is not None:
                stale.update(old.user_keys)
        for alias in stale - set(keys):
            self._by_key.pop(alias, None)
        for key in keys:
            self._by_key[key] = record
    return keys

SqliteOAuthTokenStore

基于 SQLite 的 feishu.auth.user_tokens.OAuthTokenStore 实现，用户态凭证跨重启存活。

同一用户的每个别名键写入一行、指向同一份记录 JSON，从而 open_id / union_id / user_id 任一皆可命中。

参数：

名称 类型 描述 默认

db_path

str | Path

SQLite 数据库文件路径。

 必需

示例：

Python Console Session
1
>>> store = SqliteOAuthTokenStore(":memory:")
源代码位于： feishu/auth/user_tokens.py
Python
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
class SqliteOAuthTokenStore:
    r"""
    基于 SQLite 的 [feishu.auth.user_tokens.OAuthTokenStore][] 实现，用户态凭证跨重启存活。

    同一用户的每个别名键写入一行、指向同一份记录 JSON，从而 open_id / union_id / user_id 任一皆可命中。

    Args:
        db_path: SQLite 数据库文件路径。

    Examples:
        >>> store = SqliteOAuthTokenStore(":memory:")  # doctest:+SKIP
    """

    def __init__(self, db_path: str | Path) -> None:
        self._db = _connect(db_path)
        self._db.execute(
            "CREATE TABLE IF NOT EXISTS oauth_tokens "
            "(user_key TEXT PRIMARY KEY, record TEXT NOT NULL, updated_at INTEGER)"
        )
        self._db.commit()
        self._lock = asyncio.Lock()

    async def get(self, user: Mapping[str, Any]) -> TokenRecord | None:
        r"""参见 [feishu.auth.user_tokens.OAuthTokenStore.get][]。"""
        keys = user_identity_keys(user)
        async with self._lock:
            for key in keys:
                row = self._db.execute("SELECT record FROM oauth_tokens WHERE user_key = ?", (key,)).fetchone()
                if row is not None:
                    return TokenRecord.from_dict(json.loads(row[0]))
        return None

    async def save(
        self,
        token_data: Mapping[str, Any],
        *,
        user_info: Mapping[str, Any] | None = None,
        user_keys: tuple[str, ...] = (),
    ) -> tuple[str, ...]:
        r"""参见 [feishu.auth.user_tokens.OAuthTokenStore.save][]（无用户身份时抛出 `ValueError`）。"""
        keys = _keys_for_save(user_info, user_keys)
        record = TokenRecord.from_token_data(token_data, keys, now=_now())
        payload = json.dumps(record.to_dict(), ensure_ascii=False)
        now = _now()
        async with self._lock:
            # Reconcile aliases: collect every alias of any existing row that shares a key with the
            # new identity, then delete them all and re-insert only the current key set — so a re-auth
            # with fewer aliases cannot leave a stale row pointing at the dead old record.
            placeholders = ",".join("?" * len(keys))
            rows = self._db.execute(
                f"SELECT record FROM oauth_tokens WHERE user_key IN ({placeholders})", tuple(keys)
            ).fetchall()
            stale: set[str] = set(keys)
            for (record_json,) in rows:
                stale.update(TokenRecord.from_dict(json.loads(record_json)).user_keys)
            delete_placeholders = ",".join("?" * len(stale))
            self._db.execute(f"DELETE FROM oauth_tokens WHERE user_key IN ({delete_placeholders})", tuple(stale))
            self._db.executemany(
                "INSERT INTO oauth_tokens (user_key, record, updated_at) VALUES (?, ?, ?)",
                [(key, payload, now) for key in keys],
            )
            self._db.commit()
        return keys

get async

Python
get(user: Mapping[str, Any]) -> TokenRecord | None

参见 feishu.auth.user_tokens.OAuthTokenStore.get。

源代码位于： feishu/auth/user_tokens.py
Python
263
264
265
266
267
268
269
270
271
async def get(self, user: Mapping[str, Any]) -> TokenRecord | None:
    r"""参见 [feishu.auth.user_tokens.OAuthTokenStore.get][]。"""
    keys = user_identity_keys(user)
    async with self._lock:
        for key in keys:
            row = self._db.execute("SELECT record FROM oauth_tokens WHERE user_key = ?", (key,)).fetchone()
            if row is not None:
                return TokenRecord.from_dict(json.loads(row[0]))
    return None

save async

Python
save(token_data: Mapping[str, Any], *, user_info: Mapping[str, Any] | None = None, user_keys: tuple[str, ...] = ()) -> tuple[str, ...]

参见 feishu.auth.user_tokens.OAuthTokenStore.save（无用户身份时抛出 ValueError）。

源代码位于： feishu/auth/user_tokens.py
Python
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
async def save(
    self,
    token_data: Mapping[str, Any],
    *,
    user_info: Mapping[str, Any] | None = None,
    user_keys: tuple[str, ...] = (),
) -> tuple[str, ...]:
    r"""参见 [feishu.auth.user_tokens.OAuthTokenStore.save][]（无用户身份时抛出 `ValueError`）。"""
    keys = _keys_for_save(user_info, user_keys)
    record = TokenRecord.from_token_data(token_data, keys, now=_now())
    payload = json.dumps(record.to_dict(), ensure_ascii=False)
    now = _now()
    async with self._lock:
        # Reconcile aliases: collect every alias of any existing row that shares a key with the
        # new identity, then delete them all and re-insert only the current key set — so a re-auth
        # with fewer aliases cannot leave a stale row pointing at the dead old record.
        placeholders = ",".join("?" * len(keys))
        rows = self._db.execute(
            f"SELECT record FROM oauth_tokens WHERE user_key IN ({placeholders})", tuple(keys)
        ).fetchall()
        stale: set[str] = set(keys)
        for (record_json,) in rows:
            stale.update(TokenRecord.from_dict(json.loads(record_json)).user_keys)
        delete_placeholders = ",".join("?" * len(stale))
        self._db.execute(f"DELETE FROM oauth_tokens WHERE user_key IN ({delete_placeholders})", tuple(stale))
        self._db.executemany(
            "INSERT INTO oauth_tokens (user_key, record, updated_at) VALUES (?, ?, ?)",
            [(key, payload, now) for key in keys],
        )
        self._db.commit()
    return keys

UserTokenProvider

解析并按需刷新用户态凭证，产出用户态飞书客户端。

as_user(user) 正是 feishu.agent.context.ToolContext 所需的提供方接口：传给 feishu.agent.loop.AgentEngine 的 user_tokens 后，工具即可以请求用户的身份执行读写。访问令牌将在临近过期 （refresh_skew_seconds）时用一次性轮换的 refresh_token 刷新并整体覆盖保存。

参数：

名称 类型 描述 默认

client

Any

飞书客户端，需提供 oauth（授权/刷新/用户信息）与 as_user(token)。

 必需

store

OAuthTokenStore

用户态凭证存储 feishu.auth.user_tokens.OAuthTokenStore。

 必需

refresh_skew_seconds

int

提前刷新的余量秒数。默认为 60。

 60

logger

Logger | None

日志器。

 None
源代码位于： feishu/auth/user_tokens.py
Python
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
class UserTokenProvider:
    r"""
    解析并按需刷新用户态凭证，产出用户态飞书客户端。

    `as_user(user)` 正是 [feishu.agent.context.ToolContext][] 所需的提供方接口：传给
    [feishu.agent.loop.AgentEngine][] 的 `user_tokens` 后，工具即可以请求用户的身份执行读写。访问令牌将在临近过期
    （`refresh_skew_seconds`）时用一次性轮换的 `refresh_token` 刷新并整体覆盖保存。

    Args:
        client: 飞书客户端，需提供 `oauth`（授权/刷新/用户信息）与 `as_user(token)`。
        store: 用户态凭证存储 [feishu.auth.user_tokens.OAuthTokenStore][]。
        refresh_skew_seconds: 提前刷新的余量秒数。默认为 `60`。
        logger: 日志器。
    """

    def __init__(
        self,
        client: Any,
        store: OAuthTokenStore,
        *,
        refresh_skew_seconds: int = 60,
        logger: logging.Logger | None = None,
    ) -> None:
        self.client = client
        self.store = store
        self.refresh_skew_seconds = refresh_skew_seconds
        self._log = logger or logging.getLogger("feishu")

    async def user_token(self, user: Mapping[str, Any]) -> str | None:
        r"""解析有效的用户态访问令牌；临近过期则刷新；无记录或已失效返回 `None`。"""
        record = await self._valid_record(user)
        return record.user_access_token if record is not None else None

    async def _valid_record(self, user: Mapping[str, Any]) -> TokenRecord | None:
        r"""读取一个有效用户 token 记录；临近过期时刷新，硬过期或无 token 时返回 `None`。"""
        record = await self.store.get(user)
        if record is None:
            return None
        if record.refresh_token and record.is_expiring(self.refresh_skew_seconds, _now()):
            try:
                token_data = await self.client.oauth.refresh(record.refresh_token)
                if "scope" not in token_data and record.scope:
                    token_data = {**token_data, "scope": record.scope}
                await self.store.save(token_data, user_keys=record.user_keys)
                record = await self.store.get(user)
            except Exception:  # noqa: BLE001 — fall back to the stored token; drop only if hard-expired
                self._log.warning("user_token: refresh failed; falling back to the stored token", exc_info=True)
        if record is None or record.is_expired(_now()) or not record.user_access_token:
            return None
        return record

    async def has_scopes(self, user: Mapping[str, Any], scopes: Iterable[str]) -> bool:
        r"""返回当前用户是否已有有效 token，且 token 的 `scope` 覆盖所需权限。"""
        required = {scope for scope in scopes if scope}
        if not required:
            return True
        record = await self._valid_record(user)
        if record is None:
            return False
        granted = _scope_set(record.scope)
        return required.issubset(granted)

    async def as_user(self, user: Mapping[str, Any]) -> Any | None:
        r"""产出当前用户的用户态飞书客户端；无有效凭证时返回 `None`。"""
        token = await self.user_token(user)
        if not token:
            return None
        return self.client.as_user(token)

    async def complete_authorization(self, code: str, *, redirect_uri: str | None = None) -> tuple[str, ...]:
        r"""完成 OAuth 回调：用 `code` 换取凭证、读取用户信息并保存，返回写入的别名键。"""
        token_data = await self.client.oauth.exchange_code(code, redirect_uri=redirect_uri)
        access_token = token_data.get("access_token")
        info = await self.client.oauth.user_info(access_token) if access_token else {}
        return await self.store.save(token_data, user_info=dict(info))

    def authorize_url(
        self, redirect_uri: str, *, scope: str | list[str] | tuple[str, ...] | None = None, state: str | None = None
    ) -> str:
        r"""生成授权跳转 URL（透传给 [feishu.auth.oauth.OAuthNamespace.authorize_url][]）。"""
        return self.client.oauth.authorize_url(redirect_uri, scope=scope, state=state)

user_token async

Python
user_token(user: Mapping[str, Any]) -> str | None

解析有效的用户态访问令牌；临近过期则刷新；无记录或已失效返回 None。

源代码位于： feishu/auth/user_tokens.py
Python
334
335
336
337
async def user_token(self, user: Mapping[str, Any]) -> str | None:
    r"""解析有效的用户态访问令牌；临近过期则刷新；无记录或已失效返回 `None`。"""
    record = await self._valid_record(user)
    return record.user_access_token if record is not None else None

has_scopes async

Python
has_scopes(user: Mapping[str, Any], scopes: Iterable[str]) -> bool

返回当前用户是否已有有效 token，且 token 的 scope 覆盖所需权限。

源代码位于： feishu/auth/user_tokens.py
Python
357
358
359
360
361
362
363
364
365
366
async def has_scopes(self, user: Mapping[str, Any], scopes: Iterable[str]) -> bool:
    r"""返回当前用户是否已有有效 token，且 token 的 `scope` 覆盖所需权限。"""
    required = {scope for scope in scopes if scope}
    if not required:
        return True
    record = await self._valid_record(user)
    if record is None:
        return False
    granted = _scope_set(record.scope)
    return required.issubset(granted)

as_user async

Python
as_user(user: Mapping[str, Any]) -> Any | None

产出当前用户的用户态飞书客户端；无有效凭证时返回 None。

源代码位于： feishu/auth/user_tokens.py
Python
368
369
370
371
372
373
async def as_user(self, user: Mapping[str, Any]) -> Any | None:
    r"""产出当前用户的用户态飞书客户端；无有效凭证时返回 `None`。"""
    token = await self.user_token(user)
    if not token:
        return None
    return self.client.as_user(token)

complete_authorization async

Python
complete_authorization(code: str, *, redirect_uri: str | None = None) -> tuple[str, ...]

完成 OAuth 回调：用 code 换取凭证、读取用户信息并保存，返回写入的别名键。

源代码位于： feishu/auth/user_tokens.py
Python
375
376
377
378
379
380
async def complete_authorization(self, code: str, *, redirect_uri: str | None = None) -> tuple[str, ...]:
    r"""完成 OAuth 回调：用 `code` 换取凭证、读取用户信息并保存，返回写入的别名键。"""
    token_data = await self.client.oauth.exchange_code(code, redirect_uri=redirect_uri)
    access_token = token_data.get("access_token")
    info = await self.client.oauth.user_info(access_token) if access_token else {}
    return await self.store.save(token_data, user_info=dict(info))

authorize_url

Python
authorize_url(redirect_uri: str, *, scope: str | list[str] | tuple[str, ...] | None = None, state: str | None = None) -> str

生成授权跳转 URL（透传给 feishu.auth.oauth.OAuthNamespace.authorize_url）。

源代码位于： feishu/auth/user_tokens.py
Python
382
383
384
385
386
def authorize_url(
    self, redirect_uri: str, *, scope: str | list[str] | tuple[str, ...] | None = None, state: str | None = None
) -> str:
    r"""生成授权跳转 URL（透传给 [feishu.auth.oauth.OAuthNamespace.authorize_url][]）。"""
    return self.client.oauth.authorize_url(redirect_uri, scope=scope, state=state)

user_identity_keys

Python
user_identity_keys(user: Mapping[str, Any]) -> tuple[str, ...]

把用户标识映射归一为稳定、**带类型前缀**的多别名键（如 open_id:ou_x）——全库唯一的用户身份键表示。

前缀（open_id: / union_id: / user_id:）避免不同 ID 类型的裸值互相碰撞。用于按用户隔离 token / 分享 文件 / 收款账户等存储，以及把审批绑定到发起人。逆操作见 feishu.auth.user_tokens.user_from_identity_keys。

源代码位于： feishu/auth/user_tokens.py
Python
53
54
55
56
57
58
59
60
def user_identity_keys(user: Mapping[str, Any]) -> tuple[str, ...]:
    r"""
    把用户标识映射归一为稳定、**带类型前缀**的多别名键（如 `open_id:ou_x`）——全库唯一的用户身份键表示。

    前缀（`open_id:` / `union_id:` / `user_id:`）避免不同 ID 类型的裸值互相碰撞。用于按用户隔离 token / 分享
    文件 / 收款账户等存储，以及把审批绑定到发起人。逆操作见 [feishu.auth.user_tokens.user_from_identity_keys][]。
    """
    return tuple(f"{key}:{user[key]}" for key in _USER_ID_KEYS if user.get(key))

user_from_identity_keys

Python
user_from_identity_keys(keys: Iterable[str]) -> dict[str, str]

feishu.auth.user_tokens.user_identity_keys 的逆：把带前缀的身份键还原为 {kind: value} 用户标识映射。

源代码位于： feishu/auth/user_tokens.py
Python
63
64
65
66
67
68
69
70
71
def user_from_identity_keys(keys: Iterable[str]) -> dict[str, str]:
    r"""[feishu.auth.user_tokens.user_identity_keys][] 的逆：把带前缀的身份键还原为 `{kind: value}` 用户标识映射。"""
    valid = set(_USER_ID_KEYS)
    user: dict[str, str] = {}
    for key in keys:
        kind, _, value = key.partition(":")
        if kind in valid and value and kind not in user:
            user[kind] = value
    return user