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.

Parameters:

Name Type Description Default
data str
required
event str | None
None
id str | None
None
retry int | None
None
Source code in bovine/bovine/types/__init__.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 

encode() -> bytes

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

Source code in bovine/bovine/types/__init__.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 staticmethod 

parse(raw: bytes)

Parses bytes to ServerSentEvent

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

parse_utf8 staticmethod 

parse_utf8(raw: str)

Parses utf-8 encoded string to ServerSentEvent

Source code in bovine/bovine/types/__init__.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/__init__.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"

jrd

JrdData

Bases: BaseModel

See RFC 6415

>>> JrdData(subject="acct:actor@test.example").model_dump(exclude_none=True)
{'subject': 'acct:actor@test.example'}

Parameters:

Name Type Description Default
subject str | None
None
expires datetime | None

expiration date time

 None
aliases List[str] | None

value a string array containing the values of each element in order

 None
properties Dict[str, Any] | None

value an object with each element included as a name/value pair with the value of the type attribute as name, and element value included as a string value.

 None
links List[JrdLink] | None

a single name/value pair with the name ‘links’, and value an array with each element included as an object

 None
Source code in bovine/bovine/types/jrd.py 
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
class JrdData(BaseModel):
    """
    See [RFC 6415](https://www.packetizer.com/rfc/rfc6415/)

    ```pycon
    >>> JrdData(subject="acct:actor@test.example").model_dump(exclude_none=True)
    {'subject': 'acct:actor@test.example'}

    ```
    """

    model_config = ConfigDict(
        extra="allow",
    )
    subject: str | None = Field(None, examples=["acct:actor@test.example"])
    expires: Optional[datetime] = Field(
        None,
        description="""
    expiration date time
    """,
    )
    aliases: List[str] | None = Field(
        None,
        description="""
    value a string array containing the values of each element in order
    """,
    )
    properties: Optional[Dict[str, Any]] = Field(
        None,
        description="""
    value an object with each element included as a name/value pair with the value of the type attribute as name, and element value included as a string value.
    """,
    )
    links: Optional[List[JrdLink]] = Field(
        None,
        description="""
     a single name/value pair with the name 'links', and value an array with each element included as an object
    """,
    )

Bases: BaseModel

See RFC 7033.

>>> JrdLink(rel="self", href="http://test.example/actor").model_dump(exclude_none=True)
{'rel': 'self', 'href': 'http://test.example/actor'}

Parameters:

Name Type Description Default
rel str | None

rel

 None
type str | None
None
href str | None

when used with the ‘href’ attribute, conveys a link relation between the host described by the document and a common target URI

 None
titles Dict[str, Any] | None

titles

 None
properties Dict[str, Any] | None

properties

 None
template str | None

template attribute conveys a relation whose context is an individual resource within the host-meta document scope,

 None
Source code in bovine/bovine/types/jrd.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class JrdLink(BaseModel):
    """
    See [RFC 7033](https://www.rfc-editor.org/rfc/rfc7033.html).

    ```pycon
    >>> JrdLink(rel="self", href="http://test.example/actor").model_dump(exclude_none=True)
    {'rel': 'self', 'href': 'http://test.example/actor'}

    ```
    """

    model_config = ConfigDict(
        extra="allow",
    )
    rel: str | None = Field(None, examples=["self"], description="rel")
    type: str | None = Field(None, examples=["application/activity+json"])
    href: str | None = Field(
        None,
        examples=["http://test.example/actor"],
        description="""
    when used with the 'href' attribute, conveys a link relation between the host described by the document and a common target URI
    """,
    )
    titles: Optional[Dict[str, Any]] = Field(
        None,
        description="""
    titles
    """,
    )
    properties: Optional[Dict[str, Any]] = Field(
        None,
        description="""
    properties
    """,
    )
    template: Optional[str] = Field(
        None,
        description="""
    template attribute conveys a relation whose context is an individual resource within the host-meta document scope,
    """,
    )

nodeinfo

NodeInfo

Bases: BaseModel

NodeInfo schema version 2.0.

Usage: 

>>> NodeInfo(software={"name": "bovine", "version": "0.0.1"}).model_dump()
{'version': '2.0',
    'software': {'name': 'bovine', 'version': '0.0.1'},
    'protocols': [<Protocol.activitypub: 'activitypub'>],
    'services': {'inbound': [],
        'outbound': []},
    'openRegistrations': False,
    'usage': {'users':
        {'total': 0, 'activeHalfyear': 0, 'activeMonth': 0},
        'localPosts': 0,
        'localComments': 0},
    'metadata': {}}

Parameters:

Name Type Description Default
version str

The schema version, must be 2.0.

 '2.0'
software Software

Metadata about server software in use.

 required
protocols List[Protocol]

The protocols supported on this server.

 [<Protocol.activitypub: 'activitypub'>]
services Services

The third party sites this server can connect to via their application API.

 Services(inbound=[], outbound=[])
openRegistrations bool

Whether this server allows open self-registration.

 False
usage Usage

Usage statistics for this server.

 <dynamic>
metadata Dict[str, Any]

Free form key value pairs for software specific values. Clients should not rely on any specific key present.

 {}
Source code in bovine/bovine/types/nodeinfo.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
class NodeInfo(BaseModel):
    """
    NodeInfo schema version 2.0.

    Usage:
    ```pycon
    >>> NodeInfo(software={"name": "bovine", "version": "0.0.1"}).model_dump()
    {'version': '2.0',
        'software': {'name': 'bovine', 'version': '0.0.1'},
        'protocols': [<Protocol.activitypub: 'activitypub'>],
        'services': {'inbound': [],
            'outbound': []},
        'openRegistrations': False,
        'usage': {'users':
            {'total': 0, 'activeHalfyear': 0, 'activeMonth': 0},
            'localPosts': 0,
            'localComments': 0},
        'metadata': {}}

    ```
    """

    model_config = ConfigDict(
        extra="forbid",
    )
    version: str = Field(
        "2.0",
        description="""
    The schema version, must be 2.0.
    """,
    )
    software: Software = Field(
        description="""
    Metadata about server software in use.
    """
    )
    protocols: List[Protocol] = Field(
        [Protocol.activitypub],
        min_length=1,
        description="""
    The protocols supported on this server.
    """,
    )
    services: Services = Field(
        Services(inbound=[], outbound=[]),
        description="""
    The third party sites this server can connect to via their application API.
    """,
    )
    openRegistrations: bool = Field(
        False,
        description="""
    Whether this server allows open self-registration.
    """,
    )
    usage: Usage = Field(
        default_factory=Usage,
        description="""
    Usage statistics for this server.
    """,
    )
    metadata: Dict[str, Any] = Field(
        {},
        description="""
    Free form key value pairs for software specific values. Clients should not rely on any specific key present.
    """,
    )

Services

Bases: BaseModel

The third party sites this server can connect to via their application API.

Parameters:

Name Type Description Default
inbound List[InboundEnum]

The third party sites this server can retrieve messages from for combined display with regular traffic.

 required
outbound List[OutboundEnum]

The third party sites this server can publish messages to on the behalf of a user.

 required
Source code in bovine/bovine/types/nodeinfo.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class Services(BaseModel):
    """
    The third party sites this server can connect to via their application API.
    """

    model_config = ConfigDict(
        extra="forbid",
    )
    inbound: List[InboundEnum] = Field(
        min_length=0,
        description="""
    The third party sites this server can retrieve messages from for combined display with regular traffic.
    """,
    )
    outbound: List[OutboundEnum] = Field(
        min_length=0,
        description="""
    The third party sites this server can publish messages to on the behalf of a user.
    """,
    )

Software

Bases: BaseModel

Metadata about server software in use.

Parameters:

Name Type Description Default
name str

The canonical name of this server software.

 required
version str

The version of this server software.

 required
Source code in bovine/bovine/types/nodeinfo.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Software(BaseModel):
    """
    Metadata about server software in use.
    """

    model_config = ConfigDict(
        extra="forbid",
    )
    name: constr(pattern=r"^[a-z0-9-]+$") = Field(
        description="""
    The canonical name of this server software.
    """
    )
    version: str = Field(
        description="""
    The version of this server software.
    """
    )

Usage

Bases: BaseModel

Usage statistics for this server.

Parameters:

Name Type Description Default
users Users

statistics about the users of this server.

 <dynamic>
localPosts int

The amount of posts that were made by users that are registered on this server.

 0
localComments int

The amount of comments that were made by users that are registered on this server.

 0
Source code in bovine/bovine/types/nodeinfo.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class Usage(BaseModel):
    """
    Usage statistics for this server.
    """

    model_config = ConfigDict(
        extra="forbid",
    )
    users: Users = Field(
        default_factory=Users,
        description="""
    statistics about the users of this server.
    """,
    )
    localPosts: conint(ge=0) = Field(
        0,
        description="""
    The amount of posts that were made by users that are registered on this server.
    """,
    )
    localComments: conint(ge=0) = Field(
        0,
        description="""
    The amount of comments that were made by users that are registered on this server.
    """,
    )

Users

Bases: BaseModel

statistics about the users of this server.

Parameters:

Name Type Description Default
total int

The total amount of on this server registered users.

 0
activeHalfyear int

The amount of users that signed in at least once in the last 180 days.

 0
activeMonth int

The amount of users that signed in at least once in the last 30 days.

 0
Source code in bovine/bovine/types/nodeinfo.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class Users(BaseModel):
    """
    statistics about the users of this server.
    """

    model_config = ConfigDict(
        extra="forbid",
    )
    total: conint(ge=0) = Field(
        0,
        description="""
    The total amount of on this server registered users.
    """,
    )
    activeHalfyear: conint(ge=0) = Field(
        0,
        description="""
    The amount of users that signed in at least once in the last 180 days.
    """,
    )
    activeMonth: conint(ge=0) = Field(
        0,
        description="""
    The amount of users that signed in at least once in the last 30 days.
    """,
    )