Skip to content

bovine.types

ServerSentEvent dataclass

A server sent event as defined in Event Source Interface <https://html.spec.whatwg.org/multipage/server-sent-events.html#server-sent-events>_. This is used by the event_source function of the BovineClient.

Source code in bovine/bovine/types.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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
83
84
85
86
87
88
89
90
91
@dataclass
class ServerSentEvent:
    """A server sent event as defined in
    `Event Source Interface
    <https://html.spec.whatwg.org/multipage/server-sent-events.html#server-sent-events>`_.
    This is used by the event_source function
    of the BovineClient."""

    data: str
    event: str | None = None
    id: str | None = None
    retry: int | None = None

    def encode(self) -> bytes:
        """Encodes the server sent event as bytes to send to the client."""
        message = f"data: {self.data}"
        if self.event is not None:
            message = f"{message}\nevent: {self.event}"
        if self.id is not None:
            message = f"{message}\nid: {self.id}"
        if self.retry is not None:
            message = f"{message}\nretry: {self.retry}"
        message = f"{message}\n\n"
        return message.encode("utf-8")

    @staticmethod
    def parse(raw: bytes):
        """Parses bytes to ServerSentEvent"""
        return ServerSentEvent.parse_utf8(raw.decode("utf-8"))

    @staticmethod
    def parse_utf8(raw: str):
        """Parses utf-8 encoded string to ServerSentEvent"""
        data = None
        event = None
        event_id = None
        retry = None
        for line in raw.splitlines():
            m = sse_line_pattern.match(line)
            if m is None:
                logger.warning(f"Invalid Server Sent Event line: '{line}'")
                continue

            name = m.group("name")
            if name == "":
                continue

            value = m.group("value")

            if name == "data":
                if data:
                    data = f"{data}\n{value}"
                else:
                    data = value
            elif name == "event":
                event = value
            elif name == "id":
                event_id = value
            elif name == "retry":
                retry = int(value)

        if not isinstance(data, str):
            return None

        return ServerSentEvent(data, event, event_id, retry)

encode()

Encodes the server sent event as bytes to send to the client.

Source code in bovine/bovine/types.py 
40
41
42
43
44
45
46
47
48
49
50
def encode(self) -> bytes:
    """Encodes the server sent event as bytes to send to the client."""
    message = f"data: {self.data}"
    if self.event is not None:
        message = f"{message}\nevent: {self.event}"
    if self.id is not None:
        message = f"{message}\nid: {self.id}"
    if self.retry is not None:
        message = f"{message}\nretry: {self.retry}"
    message = f"{message}\n\n"
    return message.encode("utf-8")

parse(raw) staticmethod

Parses bytes to ServerSentEvent

Source code in bovine/bovine/types.py 
52
53
54
55
@staticmethod
def parse(raw: bytes):
    """Parses bytes to ServerSentEvent"""
    return ServerSentEvent.parse_utf8(raw.decode("utf-8"))

parse_utf8(raw) staticmethod

Parses utf-8 encoded string to ServerSentEvent

Source code in bovine/bovine/types.py 
57
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
83
84
85
86
87
88
89
90
91
@staticmethod
def parse_utf8(raw: str):
    """Parses utf-8 encoded string to ServerSentEvent"""
    data = None
    event = None
    event_id = None
    retry = None
    for line in raw.splitlines():
        m = sse_line_pattern.match(line)
        if m is None:
            logger.warning(f"Invalid Server Sent Event line: '{line}'")
            continue

        name = m.group("name")
        if name == "":
            continue

        value = m.group("value")

        if name == "data":
            if data:
                data = f"{data}\n{value}"
            else:
                data = value
        elif name == "event":
            event = value
        elif name == "id":
            event_id = value
        elif name == "retry":
            retry = int(value)

    if not isinstance(data, str):
        return None

    return ServerSentEvent(data, event, event_id, retry)

Visibility

Bases: Enum

Defines the visiblity of an object. Currently only used for Actor, where endpoints are hidden from anyone by the owner.

Source code in bovine/bovine/types.py 
14
15
16
17
18
19
20
21
class Visibility(Enum):
    """Defines the visiblity of an object. Currently only used for Actor,
    where endpoints are hidden from anyone by the owner."""

    WEB = "WEB"
    PUBLIC = "PUBLIC"
    RESTRICTED = "RESTRICTED"
    OWNER = "OWNER"