Source code for nio.events.ephemeral

# -*- coding: utf-8 -*-

# Copyright © 2019 Damir Jelić <poljar@termina.org.uk>
#
# Permission to use, copy, modify, and/or distribute this software for
# any purpose with or without fee is hereby granted, provided that the
# above copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
# SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
# RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
# CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

"""nio Ephemeral events.

Ephemeral events are a special type of events that are not recorded in the room
history.

Ephemeral events are used for typing notifications and read receipts.

"""

from typing import List, Dict

from dataclasses import dataclass, field

from ..schemas import Schemas
from .misc import verify_or_none


[docs]@dataclass class EphemeralEvent: """Base class for ephemeral events."""
[docs] @classmethod @verify_or_none(Schemas.ephemeral_event) def parse_event(cls, event_dict): """Parse an ephemeral event and create a higher level event object. This function parses the type of the ephemeral event and produces a higher level event object representing the parsed event. The event structure is checked for correctness and the event fields are type-checked. If this validation process fails for an event None will be returned. If the event has an unknown type None is returned as well. Args: event_dict (dict): The dictionary representation of the event. """ if event_dict["type"] == "m.typing": return TypingNoticeEvent.from_dict(event_dict) if event_dict["type"] == "m.receipt": return ReceiptEvent.from_dict(event_dict) return None
[docs] @classmethod def from_dict(cls, parsed_dict): """Create an Ephemeral event from a dictionary. Args: parsed_dict (dict): The dictionary representation of the event. """ raise NotImplementedError()
[docs]@dataclass class TypingNoticeEvent(EphemeralEvent): """Informs the client of the list of users currently typing in a room. Attributes: users (List): The list of user IDs typing in this room, if any. """ users: List = field()
[docs] @classmethod @verify_or_none(Schemas.m_typing) def from_dict(cls, parsed_dict): return cls(parsed_dict["content"]["user_ids"])
[docs]@dataclass class Receipt: """Receipt of a user acknowledging an event. If `receipt_type` is "m.read", then it is a read receipt and shows the last event that a user has read. Attributes: event_id (str): the ID of the event being acknowleged receipt_type (str): the type of receipt being received; this is commonly "m.read" for read receipts. user_id (str): the ID of the user who is acknowledging the event. timestamp (int): The timestamp the receipt was sent at. """ event_id: str = field() receipt_type: str = field() user_id: str = field() timestamp: int = field()
[docs]@dataclass class ReceiptEvent(EphemeralEvent): """Informs the client of changes in the newest events seen by users. A ReceiptEvent can contain multiple event_ids seen by many different users. At the time of writing, all Receipts have a `receipt_type` of "m.read" and are read receipts, but this may change in the future. Attributes: receipts (List[Receipt]): The list of `Receipt`s in this event. """ receipts: List[Receipt] = field()
[docs] @classmethod @verify_or_none(Schemas.m_receipt) def from_dict(cls, parsed_dict): event_receipts: List[Receipt] = [] for event_id, event in parsed_dict["content"].items(): for receipt_type, receipt in event.items(): for user_id, user in receipt.items(): # Synapse pre-0.99.3 has a bug where it sends invalid # ts values. https://github.com/matrix-org/synapse/issues/4898 if isinstance(user, dict) and "ts" in user: event_receipts.append( Receipt(event_id, receipt_type, user_id, user["ts"]) ) return cls(event_receipts)