|
| 1 | +"""Use Spond 'get' functions to summarise available data. |
| 2 | +
|
| 3 | +Intended as a simple end-to-end test for assurance when making changes. |
| 4 | +Uses all existing group, event, message `get_` methods. |
| 5 | +Doesn't yet use `get_person(id)` or any `send_`, `update_` methods.""" |
| 6 | + |
| 7 | +import asyncio |
| 8 | +import random |
| 9 | + |
| 10 | +from config import password, username |
| 11 | +from spond import spond |
| 12 | + |
| 13 | + |
| 14 | +async def main() -> None: |
| 15 | + s = spond.Spond(username=username, password=password) |
| 16 | + |
| 17 | + print("Getting all groups...") |
| 18 | + groups = await s.get_groups() |
| 19 | + print(f"{len(groups)} groups:") |
| 20 | + for i, group in enumerate(groups): |
| 21 | + print(f"[{i}] {_group_summary(group)}") |
| 22 | + |
| 23 | + print("Getting a random group by id...") |
| 24 | + random_group_id = random.choice(groups)["id"] |
| 25 | + group = await s.get_group(random_group_id) |
| 26 | + print(f"{_group_summary(group)}") |
| 27 | + |
| 28 | + print("\nGetting up to 10 events...") |
| 29 | + events = await s.get_events(max_events=10) |
| 30 | + print(f"{len(events)} events:") |
| 31 | + for i, event in enumerate(events): |
| 32 | + print(f"[{i}] {_event_summary(event)}") |
| 33 | + |
| 34 | + print("Getting a random event by id...") |
| 35 | + random_event_id = random.choice(events)["id"] |
| 36 | + event = await s.get_event(random_event_id) |
| 37 | + print(f"{_event_summary(event)}") |
| 38 | + |
| 39 | + print("\nGetting up to 10 messages...") |
| 40 | + messages = await s.get_messages() |
| 41 | + print(f"{len(messages)} messages:") |
| 42 | + for i, message in enumerate(messages): |
| 43 | + print(f"[{i}] {_message_summary(message)}") |
| 44 | + |
| 45 | + # No `get_message(id)` function |
| 46 | + |
| 47 | + await s.clientsession.close() |
| 48 | + |
| 49 | + |
| 50 | +def _group_summary(group) -> str: |
| 51 | + return f"id: {group['id']}, " f"name: {group['name']}" |
| 52 | + |
| 53 | + |
| 54 | +def _event_summary(event) -> str: |
| 55 | + return ( |
| 56 | + f"id: {event['id']}, " |
| 57 | + f"name: {event['heading']}, " |
| 58 | + f"startTimestamp: {event['startTimestamp']}" |
| 59 | + ) |
| 60 | + |
| 61 | + |
| 62 | +def _message_summary(message) -> str: |
| 63 | + return ( |
| 64 | + f"id: {message['id']}, " |
| 65 | + f"timestamp: {message['message']['timestamp']}, " |
| 66 | + f"text: {_abbreviate(message['message']['text'], length=64)}, " |
| 67 | + ) |
| 68 | + |
| 69 | + |
| 70 | +def _abbreviate(text, length) -> str: |
| 71 | + """Abbreviate long text, normalising line endings to escape characters.""" |
| 72 | + escaped_text = repr(text) |
| 73 | + if len(text) > length: |
| 74 | + return f"{escaped_text[0:length]}[…]" |
| 75 | + return f"{escaped_text}" |
| 76 | + |
| 77 | + |
| 78 | +loop = asyncio.new_event_loop() |
| 79 | +asyncio.set_event_loop(loop) |
| 80 | +asyncio.run(main()) |
0 commit comments