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

real-time: Data classes

Browse Source 
- Related to https://github.com/status-im/status-python-sdk/issues/46
This commit is contained in:
Nick Ninov
2026-08-22 22:22:11 +03:00
parent d82a2021dc
commit 164f439bde
9 changed files with 164 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/account.md
+3 -12
View File
@@ -645,9 +645,6 @@ Listen for new incoming messages **in real time**. This method yields raw messag
```python
from status_sdk import Account
# For terminal readability only
from rich import print as rprint
from rich.pretty import Pretty
account = Account()
params = {
@@ -657,7 +654,7 @@ params = {
account.login(**params)
for msg in account.listen_messages():
rprint(Pretty(msg))
print(msg)
```
**Note**: If you receive multiple messages at once, `contacts` and `chats` will grow.
@@ -673,9 +670,6 @@ Listen for contact requests **in real time**. Both **incoming** contact requests
```python
from status_sdk import Account
# For terminal readability only
from rich import print as rprint
from rich.pretty import Pretty
account = Account()
params = {
@@ -685,7 +679,7 @@ params = {
account.login(**params)
for request in account.listen_contact_requests():
rprint(Pretty(request))
print(request)
```
Handle each type separately:
@@ -713,9 +707,6 @@ Listen for `@0x...` mentions **in real time**.
```python
from status_sdk import Account
# For terminal readability only
from rich import print as rprint
from rich.pretty import Pretty
account = Account()
params = {
@@ -725,7 +716,7 @@ params = {
account.login(**params)
for mention in account.listen_message_mentions():
rprint(Pretty(mention))
print(mention)
```
#### `add_contact(public_key, display_name=None)`
docs/community.md
+6 -6
View File
@@ -534,11 +534,11 @@ community.delete_channel("announcements")
Listen for join requests to the community **in real time**.
Returns a `Generator` that yields one `dict` per request event:
Returns a `Generator` that yields one `models.CommunityRequest` **dataclass** per request event, so the fields are reached as attributes (`request.state`) rather than dictionary keys:
| Key | Type | Description |
| Attribute | Type | Description |
|----|----|-------------|
| `request_id` | `str` | The join request id. Pass this to [`accept`](./community.md#acceptpending_request_id) or [`decline`](./community.md#declinepending_request_id). |
| `id` | `str` | The join request id. Pass this to [`accept`](./community.md#acceptpending_request_id) or [`decline`](./community.md#declinepending_request_id). |
| `state` | `str` | The state the request moved into - see the table below. |
| `public_key` | `str` | Public key of the requesting member. |
@@ -568,12 +568,12 @@ community = Community(account, url=url)
# Auto-accept everyone who asks to join
for request in community.listen_requests():
print(f"{request['public_key']}\t{request['state']}")
print(f"{request.public_key}\t{request.state}")
if request["state"] != "pending":
if request.state != "pending":
continue
community.accept(request["request_id"])
community.accept(request.id)
community["general"].send_message("Welcome to the community!")
```
examples/agents/main.py
+18 -31
View File
@@ -91,40 +91,27 @@ if __name__ == "__main__":
)
for message in status_toolkit.account.listen_messages():
content = None
for chat in message["event"]["chats"]:
latest_message: dict = chat.get("lastMessage", {})
if not latest_message:
continue
from_public_key = latest_message.get("from")
if from_public_key != PUBLIC_KEY:
continue
content = chat["lastMessage"]["text"]
payment_requests: list[dict] = latest_message.get("paymentRequests", [])
if payment_requests:
payment_request = payment_requests[0]
amount = status_toolkit.normalize_amount(payment_request["amount"], payment_request["tokenKey"])
chain_id, token_address = payment_request["tokenKey"].split("-")
payment_content = {
"Receiver Wallet": payment_request['receiver'],
"Token Symbol": payment_request['symbol'],
"Token Address": token_address,
"Amount": amount,
"Chain ID": chain_id
}
content += f"\n---\n# Payment request\n" + "\n".join([
f"{name}: {value}"
for name, value in payment_content.items()
])
break
if not content:
if message.from_public_key != PUBLIC_KEY:
continue
content = message.content
payment_requests = message.payment_requests
if payment_requests:
payment_request = payment_requests[0]
amount = status_toolkit.normalize_amount(payment_request.amount, f"{payment_request.chain_id}-{payment_request.token_address}")
payment_content = {
"Receiver Wallet": payment_request.to_address,
"Token Symbol": payment_request.token_symbol,
"Token Address": payment_request.token_address,
"Amount": amount,
"Chain ID": payment_request.chain_id
}
content += f"\n---\n# Payment request\n" + "\n".join([
f"{name}: {value}"
for name, value in payment_content.items()
])
result = agent.invoke({
"messages": [
{
examples/community-greet/main.py
+6 -6
View File
@@ -111,17 +111,17 @@ def main(channel_name: str, approve: bool):
account.logger.info(f"Listening for incoming {community.name} [{community.id}] requests")
pending_requests = []
for request in community.listen_requests():
member_public_key: str = request["public_key"]
request_id: str = request["request_id"]
if request["state"] == "pending" and member_public_key not in pending_requests:
member_public_key: str = request.public_key
if request.state == "pending" and member_public_key not in pending_requests:
pending_requests.append(member_public_key)
if approve and request["state"] == "pending":
community.accept(request_id)
if approve and request.state == "pending":
community.accept(request.id)
account.logger.info(f"Accepted {member_public_key}")
continue
if request["state"] != "accept" or member_public_key not in pending_requests:
if request.state != "accept" or member_public_key not in pending_requests:
continue
message = generate_message(member_public_key)
examples/group-chat-moderator/main.py
+13 -15
View File
@@ -1,5 +1,5 @@
from dotenv import load_dotenv
from status_sdk import Account, GroupChat, launch_docker_container
from status_sdk import Account, GroupChat, launch_docker_container, models
from detoxify import Detoxify
import os, threading, torch
@@ -7,16 +7,16 @@ import os, threading, torch
# read-modify-write of a member's warning count must be atomic
warnings_lock = threading.Lock()
def check_message(account: Account, message: dict, warnings: dict, group_chat: GroupChat, model: Detoxify, threshold: float = 0.6, warning_limit: int = 3):
def check_message(account: Account, message: models.Message, warnings: dict, group_chat: GroupChat, model: Detoxify, threshold: float = 0.6, warning_limit: int = 3):
"""
Score a single message and warn (or remove) its author.
"""
public_key = message["from"]
public_key = message.from_public_key
if public_key == account.info["public_key"]:
return
label, score = max(model.predict(message["text"]).items(), key=lambda item: item[1])
account.logger.info(f"Message: '{message['text']}'\t\t{label} - {(score * 100):.2f}%")
label, score = max(model.predict(message.content).items(), key=lambda item: item[1])
account.logger.info(f"Message: '{message.content}'\t\t{label} - {(score * 100):.2f}%")
if score < threshold:
return
@@ -26,7 +26,7 @@ def check_message(account: Account, message: dict, warnings: dict, group_chat: G
count = warnings[public_key]
if count < warning_limit:
group_chat.send_message(f"Warning {count} /{warning_limit} - @{public_key} please keep it civil.", message["id"])
group_chat.send_message(f"Warning {count} /{warning_limit} - @{public_key} please keep it civil.", message.id)
account.logger.info(f"Sent warning to {public_key}")
elif count >= warning_limit:
group_chat.send_message(f"Removing @{public_key} member after {warning_limit} warnings.")
@@ -50,15 +50,13 @@ def main():
model = Detoxify("original", device=device)
account.logger.info(f"Listening Group Chat {group_chat.name}")
for message in account.listen_messages():
for chat in message["event"]["chats"]:
if chat["id"] != group_chat.id:
continue
threading.Thread(
target=check_message,
args=(account, chat["lastMessage"], warnings, group_chat, model),
daemon=True
).start()
if message.chat_id != group_chat.id:
continue
threading.Thread(
target=check_message,
args=(account, message, warnings, group_chat, model),
daemon=True
).start()
if __name__ == "__main__":
main()
status_sdk/__init__.py
+1 -1
View File
@@ -4,7 +4,7 @@ from .account import Account
from .group_chat import GroupChat
from .community.base import Community
from .utils import launch_docker_container
from . import exceptions
from . import exceptions, models
try:
__version__ = _version("status-sdk")
status_sdk/account.py
+15 -10
View File
@@ -8,7 +8,7 @@ from io import BytesIO
from PIL import Image
from PIL.JpegImagePlugin import JpegImageFile
from PIL.PngImagePlugin import PngImageFile
from . import constants
from . import constants, models
from .signal import Signal
from .logger import Logger
@@ -529,6 +529,7 @@ class Account:
contacts = [
{"type": "contact", "id": contact["chat_id"], "name": contact["display_name"]}
for contact in self.contacts.values()
if contact["mutual"]
]
# Group chats in RPC endpoint are chat type 3
@@ -827,7 +828,7 @@ class Account:
return not any(errors) if errors else False
def listen_contact_requests(self) -> Generator:
def listen_contact_requests(self) -> Generator[models.ContactRequest, None, None]:
"""
Listen for incoming contact requests and for contact requests that were accepted. Can be used for real time processing.
"""
@@ -838,14 +839,17 @@ class Account:
if message["type"] == "local-notifications":
category = event.get("category")
if category == "contactRequest":
message["request_type"] = "incoming"
yield message
yield models.ContactRequest(message["event"]["body"]["message"]["from"], incoming=True)
if message["type"] == "messages.new" and accepted_contact_request.search(str(message)):
message["request_type"] = "accepted"
yield message
for public_key in set(accepted_contact_request.findall(str(message))):
def listen_message_mentions(self) -> Generator:
if self.info["public_key"] == public_key:
continue
yield models.ContactRequest(public_key, accepted=True)
def listen_message_mentions(self) -> Generator[models.Message, None, None]:
"""
Listen for `@0xpublic-key` mentions. Can be used for real time processing.
"""
@@ -860,16 +864,17 @@ class Account:
current_text: str = event["body"]["message"]["text"]
if mention_everyone in current_text or account_mention in current_text:
yield message
yield models.Message.from_raw(event["body"]["message"])
def listen_messages(self) -> Generator:
def listen_messages(self) -> Generator[models.Message, None, None]:
"""
Listen for new **RAW** messages continuously. Can be used for real time processing.
"""
for message in self.signal.listen("messages.new"):
event: dict = message.get("event", {})
if "chats" in event or "messages" in event:
yield message
for raw in event["messages"]:
yield models.Message.from_raw(raw)
def get_messages(self, chat_id: str, start_timestamp: Optional[Union[str, datetime.datetime, datetime.date, pd.Timestamp]] = None, end_timestamp: Optional[Union[str, datetime.datetime, datetime.date, pd.Timestamp]] = None) -> list[dict]:
"""
status_sdk/community/base.py
+3 -7
View File
@@ -1,5 +1,5 @@
from ..account import Account
from .. import exceptions
from .. import exceptions, models
from .channel import Channel
from typing import Union, Optional, Generator
import pandas as pd
@@ -281,7 +281,7 @@ class Community:
params = [self.id, channel.id.replace(self.id, "")]
self.__account._call_rpc("messaging", "deleteCommunityChat", params)
def listen_requests(self) -> Generator:
def listen_requests(self) -> Generator[models.CommunityRequest, None, None]:
"""
Listen for commnunity requests
"""
@@ -300,11 +300,7 @@ class Community:
if not state:
continue
yield {
"request_id": request["id"],
"state": state,
"public_key": request["publicKey"]
}
yield models.CommunityRequest(request["id"], state, request["publicKey"])
@property
def categories(self) -> dict[str, str]:
status_sdk/models.py
+99
View File
@@ -0,0 +1,99 @@
from dataclasses import dataclass, field
from typing import Self, Optional
import datetime
@dataclass
class ContactRequest:
public_key: str
incoming: bool = False
accepted: bool = False
def __post_init__(self):
if not (self.incoming or self.accepted):
raise ValueError("A ContactRequest must be `incoming` or `accepted`")
@dataclass
class PaymentRequest:
to_address: str
token_symbol: str
token_address: str
chain_id: int
amount: str
@classmethod
def from_raw(cls, raw: dict) -> Self:
chain_id, token_address = raw["tokenKey"].split("-")
params = {
"to_address": raw["receiver"],
"token_symbol": raw["symbol"],
"token_address": token_address,
"chain_id": int(chain_id),
"amount": raw["amount"]
}
return cls(**params)
@dataclass
class Message:
id: str
chat_id: str
content: str
content_type: str
from_public_key: str
timestamp: datetime.datetime
chat_type: str
reply_id: Optional[str] = None
payment_requests: list[PaymentRequest] = field(default_factory=list)
@classmethod
def from_raw(cls, raw: dict) -> Self:
content_type: int = raw["contentType"]
msg_type: int = raw["messageType"]
params = {
"id": raw["id"],
"chat_id": raw["chatId"],
"from_public_key": raw["from"],
"timestamp": datetime.datetime.fromtimestamp(raw["whisperTimestamp"] / 1_000)
}
if len(raw["responseTo"]) > 0:
params["reply_id"] = raw["responseTo"]
if msg_type == 5:
params["chat_type"] = "community"
elif msg_type == 1:
params["chat_type"] = "private"
elif msg_type in [2, 3]:
params["chat_type"] = "group"
# Text & Emojis
if content_type in [1, 4]:
params["content"] = raw["text"]
params["content_type"] = "text" if content_type == 1 else "image"
# Sticker
elif content_type == 2:
params["content"] = raw["sticker"]["url"]
params["content_type"] = "sticker"
# Image
elif content_type == 7:
img_path = raw["image"]
text = raw["text"]
caption = f"{text}\n\n" if len(text) > 0 else ""
params["content"] = f"{caption}{img_path}"
params["content_type"] = "image"
payments = raw.get("paymentRequests", [])
if payments:
params["payment_requests"] = [
PaymentRequest.from_raw(payment)
for payment in payments
]
return cls(**params)
@dataclass
class CommunityRequest:
id: str
state: str
public_key: str