status-im/status-python-sdk
1
0
Fork 0
mirror of https://github.com/status-im/status-python-sdk.git synced 2026-08-30 13:41:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

bot: Fetch all signals

Browse Source 
- Related to https://github.com/status-im/status-python-sdk/issues/41
This commit is contained in:
Nick Ninov
2026-08-18 19:39:24 +03:00
parent 50256ee528
commit 76ed99f040
1 changed files with 10 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
status_sdk/signal.py
+10 -2
View File
@@ -102,13 +102,18 @@ class Signal:
Listen continuously for Status Backend signals
"""
signal: dict = json.loads(signal)
if signal["type"] != self.__signal_type:
if isinstance(self.__signal_type, str) and self.__signal_type != signal["type"]:
return
if isinstance(self.__signal_type, list) and signal["type"] not in self.__signal_type:
return
event: Optional[dict] = signal.get("event", {}) or {}
data = {
"timestamp": datetime.datetime.fromtimestamp(signal["timestamp"]),
"is_error": not isinstance(event.get("error"), type(None)),
"type": signal["type"],
"error_message": event.get("error"),
"event": event
}
@@ -124,13 +129,16 @@ class Signal:
if self.__thread.is_alive():
self.__thread.join(1)
def listen(self, signal_type: str):
def listen(self, signal_type: Optional[Union[str, list[str]]] = None):
"""
Listen for a specific Signal forever
Parameters:
- `signal_type` - the "type" as it is in Status Backend
"""
if isinstance(signal_type, list) and len(signal_type) == 0:
signal_type = None
self.__signal_type = signal_type
ws = websocket.WebSocketApp(
self.__url,