跳转至

notifications

feishu.gateway.notifications

EventMessageStore

Bases: Protocol

Stores the Feishu message ID associated with an event ID.

源代码位于： feishu/gateway/notifications.py
Python
33
34
35
36
37
38
39
40
41
42
class EventMessageStore(Protocol):
    r"""Stores the Feishu message ID associated with an event ID."""

    def get(self, event_id: str) -> str | None:
        r"""Return the Feishu message ID previously sent for ``event_id``."""
        ...

    def set(self, event_id: str, message_id: str) -> None:
        r"""Persist the Feishu message ID for ``event_id``."""
        ...

get

Python
get(event_id: str) -> str | None

Return the Feishu message ID previously sent for event_id.

源代码位于： feishu/gateway/notifications.py
Python
36
37
38
def get(self, event_id: str) -> str | None:
    r"""Return the Feishu message ID previously sent for ``event_id``."""
    ...

set

Python
set(event_id: str, message_id: str) -> None

Persist the Feishu message ID for event_id.

源代码位于： feishu/gateway/notifications.py
Python
40
41
42
def set(self, event_id: str, message_id: str) -> None:
    r"""Persist the Feishu message ID for ``event_id``."""
    ...

InMemoryEventMessageStore

Process-local event_id -> message_id store.

源代码位于： feishu/gateway/notifications.py
Python
45
46
47
48
49
50
51
52
53
54
55
class InMemoryEventMessageStore:
    r"""Process-local ``event_id -> message_id`` store."""

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

    def get(self, event_id: str) -> str | None:
        return self._messages.get(event_id)

    def set(self, event_id: str, message_id: str) -> None:
        self._messages[event_id] = message_id

JsonFileEventMessageStore

Small JSON-file store for event_id -> message_id state.

源代码位于： feishu/gateway/notifications.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
class JsonFileEventMessageStore:
    r"""Small JSON-file store for ``event_id -> message_id`` state."""

    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self._lock = threading.RLock()

    def get(self, event_id: str) -> str | None:
        return self._read().get(event_id)

    def set(self, event_id: str, message_id: str) -> None:
        with self._lock:
            data = self._read()
            data[event_id] = message_id
            self.path.parent.mkdir(parents=True, exist_ok=True)
            self.path.write_text(json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + "\n")

    def _read(self) -> dict[str, str]:
        with self._lock:
            if not self.path.exists():
                return {}
            data = json.loads(self.path.read_text())
            if not isinstance(data, dict):
                return {}
            return {str(key): str(value) for key, value in data.items() if isinstance(value, str)}

InteractiveCardDelivery dataclass

Result of creating or updating an event-keyed interactive card.

源代码位于： feishu/gateway/notifications.py
Python
85
86
87
88
89
90
91
@dataclass(frozen=True)
class InteractiveCardDelivery:
    r"""Result of creating or updating an event-keyed interactive card."""

    action: str
    event_id: str
    message_id: str | None

deterministic_uuid

Python
deterministic_uuid(event_id: str, *, prefix: str = 'event-') -> str

Return a deterministic Feishu idempotency UUID for an event.

源代码位于： feishu/gateway/notifications.py
Python
94
95
96
def deterministic_uuid(event_id: str, *, prefix: str = "event-") -> str:
    r"""Return a deterministic Feishu idempotency UUID for an event."""
    return prefix + hashlib.sha256(event_id.encode()).hexdigest()[:32]

upsert_interactive_card async

Python
upsert_interactive_card(client: Any, event_id: str, card: dict[str, Any], receive_id: str, *, store: EventMessageStore, receive_id_type: str = 'chat_id', uuid_prefix: str = 'event-') -> InteractiveCardDelivery

Create an interactive card, or patch the prior card for the same event.

The store provides durable event_id -> message_id state. A repeat event patches the previously delivered Feishu card rather than posting a duplicate message.

源代码位于： feishu/gateway/notifications.py
Python
 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
async def upsert_interactive_card(
    client: Any,
    event_id: str,
    card: dict[str, Any],
    receive_id: str,
    *,
    store: EventMessageStore,
    receive_id_type: str = "chat_id",
    uuid_prefix: str = "event-",
) -> InteractiveCardDelivery:
    r"""Create an interactive card, or patch the prior card for the same event.

    The ``store`` provides durable ``event_id -> message_id`` state. A repeat
    event patches the previously delivered Feishu card rather than posting a
    duplicate message.
    """
    message_id = store.get(event_id)
    if message_id:
        data = await client.im.patch(message_id, card)
        return InteractiveCardDelivery("updated", event_id, _message_id(data) or message_id)

    data = await client.im.send(
        receive_id,
        card,
        receive_id_type=receive_id_type,
        msg_type="interactive",
        uuid=deterministic_uuid(event_id, prefix=uuid_prefix),
    )
    message_id = _message_id(data)
    if message_id:
        store.set(event_id, message_id)
    return InteractiveCardDelivery("created", event_id, message_id)