跳转至

idempotency

feishu.events.idempotency

SeenStore

Bases: Protocol

事件去重存储的协议接口。

飞书在投递超时时会重试推送，导致同一 event_id 多次到达。实现本协议即可为接收器 （create_event_route、 create_card_route）或 EventDispatcher 提供幂等保证。 通过 InMemorySeenStore 即得到一个内置实现， 生产环境可改用基于 Redis 等共享存储的实现。

本协议使用 runtime_checkable，可用 isinstance 进行结构化校验；结构化校验仅要求实现 seen / mark 两个基础方法。

实现可以额外提供原子的 claim（见下）作为快路径。当存储支持时， claim 会优先调用它，从而在并发重复投递下原子地完成 「检查并标记」；未提供时则回退到 seen() + mark() 两步。

飞书文档

接收事件

源代码位于： feishu/events/idempotency.py
Python
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@runtime_checkable
class SeenStore(Protocol):
    r"""
    事件去重存储的协议接口。

    飞书在投递超时时会重试推送，导致同一 `event_id` 多次到达。实现本协议即可为接收器
    （[create_event_route][feishu.events.receiver.create_event_route]、
    [create_card_route][feishu.events.receiver.create_card_route]）或
    [EventDispatcher][feishu.events.dispatcher.EventDispatcher] 提供幂等保证。
    通过 [InMemorySeenStore][feishu.events.idempotency.InMemorySeenStore] 即得到一个内置实现，
    生产环境可改用基于 Redis 等共享存储的实现。

    本协议使用 `runtime_checkable`，可用 `isinstance` 进行结构化校验；结构化校验仅要求实现
    `seen` / `mark` 两个基础方法。

    实现可以额外提供原子的 `claim`（见下）作为快路径。当存储支持时，
    [claim][feishu.events.idempotency.claim] 会优先调用它，从而在并发重复投递下原子地完成
    「检查并标记」；未提供时则回退到 `seen()` + `mark()` 两步。

    飞书文档:
        [接收事件](https://open.feishu.cn/document/server-docs/event-subscription-guide/event-subscription-configure-/request-url-configuration-case)
    """

    async def seen(self, event_id: str) -> bool:
        r"""
        查询 `event_id` 是否已被处理过。
        """
        ...

    async def mark(self, event_id: str) -> None:
        r"""
        将 `event_id` 标记为已处理。
        """
        ...

seen async

Python
seen(event_id: str) -> bool

查询 event_id 是否已被处理过。

源代码位于： feishu/events/idempotency.py
Python
57
58
59
60
61
async def seen(self, event_id: str) -> bool:
    r"""
    查询 `event_id` 是否已被处理过。
    """
    ...

mark async

Python
mark(event_id: str) -> None

将 event_id 标记为已处理。

源代码位于： feishu/events/idempotency.py
Python
63
64
65
66
67
async def mark(self, event_id: str) -> None:
    r"""
    将 `event_id` 标记为已处理。
    """
    ...

InMemorySeenStore

基于进程内存、带 TTL 的 SeenStore 实现。

将已处理的 event_id 连同过期时间存入字典，每次访问时清理过期项。适合单进程部署与测试； 多副本部署时各进程内存互不共享，应改用基于共享存储的实现。

参数：

名称 类型 描述 默认

ttl

float

记录的存活时长（秒），超过后视为未见过。默认 3600。

 3600.0

now

Callable[[], float]

单调时钟函数，默认 time.monotonic，可注入以便测试。

 monotonic

示例：

Python Console Session
1
2
3
4
5
6
7
>>> import asyncio
>>> store = InMemorySeenStore()
>>> asyncio.run(store.seen("evt_1"))
False
>>> asyncio.run(store.mark("evt_1"))
>>> asyncio.run(store.seen("evt_1"))
True
源代码位于： feishu/events/idempotency.py
Python
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 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
class InMemorySeenStore:
    r"""
    基于进程内存、带 TTL 的 [SeenStore][feishu.events.idempotency.SeenStore] 实现。

    将已处理的 `event_id` 连同过期时间存入字典，每次访问时清理过期项。适合单进程部署与测试；
    多副本部署时各进程内存互不共享，应改用基于共享存储的实现。

    Args:
        ttl: 记录的存活时长（秒），超过后视为未见过。默认 `3600`。
        now: 单调时钟函数，默认 `time.monotonic`，可注入以便测试。

    Examples:
        >>> import asyncio
        >>> store = InMemorySeenStore()
        >>> asyncio.run(store.seen("evt_1"))
        False
        >>> asyncio.run(store.mark("evt_1"))
        >>> asyncio.run(store.seen("evt_1"))
        True
    """

    def __init__(self, ttl: float = 3600.0, *, now: Callable[[], float] = time.monotonic) -> None:
        self._ttl = ttl
        self._now = now
        self._store: dict[str, float] = {}
        self._lock = asyncio.Lock()

    async def claim(self, event_id: str) -> bool:
        r"""
        原子地认领 `event_id`：此前未标记（或已过期）则标记并返回 `True`，否则返回 `False`。

        将 `seen` 检查与 `mark` 标记合并在同一把锁内完成，消除二者之间的检查-标记竞态——
        并发投递的重复事件中只有一个会得到 `True`。

        Args:
            event_id: 待认领的事件标识。

        Returns:
            首次认领返回 `True`（应处理），重复返回 `False`（应跳过）。
        """
        async with self._lock:
            self._purge()
            if event_id in self._store:
                return False
            self._store[event_id] = self._now() + self._ttl
            return True

    async def seen(self, event_id: str) -> bool:
        r"""
        查询 `event_id` 是否在 TTL 内被标记过。

        Args:
            event_id: 待查询的事件标识。

        Returns:
            已标记且未过期返回 `True`，否则返回 `False`。
        """
        async with self._lock:
            self._purge()
            return event_id in self._store

    async def mark(self, event_id: str) -> None:
        r"""
        标记 `event_id` 为已处理，并按 TTL 设置过期时间。

        Args:
            event_id: 待标记的事件标识。
        """
        async with self._lock:
            self._purge()
            self._store[event_id] = self._now() + self._ttl

    def _purge(self) -> None:
        now = self._now()
        expired = [k for k, exp in self._store.items() if exp <= now]
        for k in expired:
            del self._store[k]

claim async

Python
claim(event_id: str) -> bool

原子地认领 event_id：此前未标记（或已过期）则标记并返回 True，否则返回 False。

将 seen 检查与 mark 标记合并在同一把锁内完成，消除二者之间的检查-标记竞态—— 并发投递的重复事件中只有一个会得到 True。

参数：

名称 类型 描述 默认
event_id
str

待认领的事件标识。

 必需

返回：

类型 描述
bool

首次认领返回 True（应处理），重复返回 False（应跳过）。
源代码位于： feishu/events/idempotency.py
Python
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def claim(self, event_id: str) -> bool:
    r"""
    原子地认领 `event_id`：此前未标记（或已过期）则标记并返回 `True`，否则返回 `False`。

    将 `seen` 检查与 `mark` 标记合并在同一把锁内完成，消除二者之间的检查-标记竞态——
    并发投递的重复事件中只有一个会得到 `True`。

    Args:
        event_id: 待认领的事件标识。

    Returns:
        首次认领返回 `True`（应处理），重复返回 `False`（应跳过）。
    """
    async with self._lock:
        self._purge()
        if event_id in self._store:
            return False
        self._store[event_id] = self._now() + self._ttl
        return True

seen async

Python
seen(event_id: str) -> bool

查询 event_id 是否在 TTL 内被标记过。

参数：

名称 类型 描述 默认
event_id
str

待查询的事件标识。

 必需

返回：

类型 描述
bool

已标记且未过期返回 True，否则返回 False。
源代码位于： feishu/events/idempotency.py
Python
117
118
119
120
121
122
123
124
125
126
127
128
129
async def seen(self, event_id: str) -> bool:
    r"""
    查询 `event_id` 是否在 TTL 内被标记过。

    Args:
        event_id: 待查询的事件标识。

    Returns:
        已标记且未过期返回 `True`，否则返回 `False`。
    """
    async with self._lock:
        self._purge()
        return event_id in self._store

mark async

Python
mark(event_id: str) -> None

标记 event_id 为已处理，并按 TTL 设置过期时间。

参数：

名称 类型 描述 默认
event_id
str

待标记的事件标识。

 必需
源代码位于： feishu/events/idempotency.py
Python
131
132
133
134
135
136
137
138
139
140
async def mark(self, event_id: str) -> None:
    r"""
    标记 `event_id` 为已处理，并按 TTL 设置过期时间。

    Args:
        event_id: 待标记的事件标识。
    """
    async with self._lock:
        self._purge()
        self._store[event_id] = self._now() + self._ttl

SqliteSeenStore

基于 SQLite、带 TTL 的 SeenStore 实现。

每个操作在 worker thread 中执行一条小型 SQLite 事务，适合生产机器人把事件去重接入已有的 SQLite 状态库。

参数：

名称 类型 描述 默认

path

str | PathLike[str]

SQLite 数据库路径。

 必需

ttl

float

记录的存活时长（秒），超过后视为未见过。默认 7 天。

 7 * 24 * 3600

now

Callable[[], float]

wall-clock 时间函数，默认 time.time；使用 wall-clock 是为了跨进程重启仍可判断过期。

 time
源代码位于： feishu/events/idempotency.py
Python
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
class SqliteSeenStore:
    r"""
    基于 SQLite、带 TTL 的 [SeenStore][feishu.events.idempotency.SeenStore] 实现。

    每个操作在 worker thread 中执行一条小型 SQLite 事务，适合生产机器人把事件去重接入已有的 SQLite
    状态库。

    Args:
        path: SQLite 数据库路径。
        ttl: 记录的存活时长（秒），超过后视为未见过。默认 7 天。
        now: wall-clock 时间函数，默认 `time.time`；使用 wall-clock 是为了跨进程重启仍可判断过期。
    """

    def __init__(
        self, path: str | os.PathLike[str], ttl: float = 7 * 24 * 3600, *, now: Callable[[], float] = time.time
    ) -> None:
        self._path = Path(path)
        self._ttl = ttl
        self._now = now

    async def claim(self, event_id: str) -> bool:
        r"""原子地认领 `event_id`：此前未标记（或已过期）则标记并返回 `True`。"""
        return bool(await asyncio.to_thread(self._claim_sync, event_id, self._now()))

    async def seen(self, event_id: str) -> bool:
        r"""查询 `event_id` 是否在 TTL 内被标记过。"""
        return bool(await asyncio.to_thread(self._seen_sync, event_id, self._now()))

    async def mark(self, event_id: str) -> None:
        r"""标记 `event_id` 为已处理，并按 TTL 设置过期时间。"""
        await asyncio.to_thread(self._mark_sync, event_id, self._now())

    def _connect(self) -> sqlite3.Connection:
        if self._path.parent != Path("."):
            self._path.parent.mkdir(parents=True, exist_ok=True)
        db = sqlite3.connect(self._path, timeout=30)
        db.execute("PRAGMA journal_mode=WAL")
        db.execute("CREATE TABLE IF NOT EXISTS seen_events (event_id TEXT PRIMARY KEY, expires_at REAL NOT NULL)")
        return db

    def _purge_expired(self, db: sqlite3.Connection, now: float) -> None:
        db.execute("DELETE FROM seen_events WHERE expires_at <= ?", (now,))

    def _claim_sync(self, event_id: str, now: float) -> bool:
        with closing(self._connect()) as db, db:
            self._purge_expired(db, now)
            try:
                db.execute(
                    "INSERT INTO seen_events (event_id, expires_at) VALUES (?, ?)",
                    (event_id, now + self._ttl),
                )
            except sqlite3.IntegrityError:
                return False
            return True

    def _seen_sync(self, event_id: str, now: float) -> bool:
        with closing(self._connect()) as db, db:
            self._purge_expired(db, now)
            row = db.execute("SELECT 1 FROM seen_events WHERE event_id = ?", (event_id,)).fetchone()
            return row is not None

    def _mark_sync(self, event_id: str, now: float) -> None:
        with closing(self._connect()) as db, db:
            self._purge_expired(db, now)
            db.execute(
                "INSERT OR REPLACE INTO seen_events (event_id, expires_at) VALUES (?, ?)",
                (event_id, now + self._ttl),
            )

claim async

Python
claim(event_id: str) -> bool

原子地认领 event_id：此前未标记（或已过期）则标记并返回 True。

源代码位于： feishu/events/idempotency.py
Python
169
170
171
async def claim(self, event_id: str) -> bool:
    r"""原子地认领 `event_id`：此前未标记（或已过期）则标记并返回 `True`。"""
    return bool(await asyncio.to_thread(self._claim_sync, event_id, self._now()))

seen async

Python
seen(event_id: str) -> bool

查询 event_id 是否在 TTL 内被标记过。

源代码位于： feishu/events/idempotency.py
Python
173
174
175
async def seen(self, event_id: str) -> bool:
    r"""查询 `event_id` 是否在 TTL 内被标记过。"""
    return bool(await asyncio.to_thread(self._seen_sync, event_id, self._now()))

mark async

Python
mark(event_id: str) -> None

标记 event_id 为已处理，并按 TTL 设置过期时间。

源代码位于： feishu/events/idempotency.py
Python
177
178
179
async def mark(self, event_id: str) -> None:
    r"""标记 `event_id` 为已处理，并按 TTL 设置过期时间。"""
    await asyncio.to_thread(self._mark_sync, event_id, self._now())

claim async

Python
claim(store: SeenStore, event_id: str) -> bool

向 store 认领 event_id，返回是否为首次见到（应处理）。

若 store 提供原子的 claim(event_id) -> bool（如 InMemorySeenStore）则优先使用，从而在并发重复投递下 也能保证「检查并标记」原子完成；否则回退到 seen() + mark() 两步（语义不变，其原子性由具体存储自行保证）。

参数：

名称 类型 描述 默认

store

SeenStore

事件去重存储。

 必需

event_id

str

待认领的事件标识。

 必需

返回：

类型 描述
bool

首次见到返回 True（应处理该事件），重复返回 False（应跳过）。
源代码位于： feishu/events/idempotency.py
Python
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
async def claim(store: SeenStore, event_id: str) -> bool:
    r"""
    向 `store` 认领 `event_id`，返回是否为首次见到（应处理）。

    若 `store` 提供原子的 `claim(event_id) -> bool`（如
    [InMemorySeenStore][feishu.events.idempotency.InMemorySeenStore]）则优先使用，从而在并发重复投递下
    也能保证「检查并标记」原子完成；否则回退到 `seen()` + `mark()` 两步（语义不变，其原子性由具体存储自行保证）。

    Args:
        store: 事件去重存储。
        event_id: 待认领的事件标识。

    Returns:
        首次见到返回 `True`（应处理该事件），重复返回 `False`（应跳过）。
    """
    store_claim = getattr(store, "claim", None)
    if callable(store_claim):
        return bool(await store_claim(event_id))
    if await store.seen(event_id):
        return False
    await store.mark(event_id)
    return True