Skip to content

bovine.activitystreams.object_factory

Object dataclass

A dataclass representing an ActivityStreams Object <https://www.w3.org/TR/activitystreams-vocabulary/#object-types>_

Source code in bovine/bovine/activitystreams/object_factory.py 
 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
 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
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@dataclass
class Object:
    """A dataclass representing an `ActivityStreams Object
    <https://www.w3.org/TR/activitystreams-vocabulary/#object-types>`_"""

    type: str
    attributed_to: Optional[str] = None
    followers: Optional[str] = None
    id: Optional[str] = None
    published: Optional[str] = None
    to: Set[str] = field(default_factory=set)
    cc: Set[str] = field(default_factory=set)

    name: Optional[str] = None
    summary: Optional[str] = None
    content: Optional[str] = None
    source: Optional[dict] = None

    width: Optional[int] = None
    height: Optional[int] = None

    in_reply_to: Optional[str] = None
    url: Optional[str] = None
    tag: List[dict] = field(default_factory=list)
    attachment: List[dict] = field(default_factory=list)
    href: Optional[str] = None
    icon: Optional[dict] = None
    media_type: Optional[str] = None

    def as_public(self):
        """makes the object public, i.e. public in to and followers in cc"""
        self.to.add("https://www.w3.org/ns/activitystreams#Public")
        if self.followers:
            self.cc.add(self.followers)
        return self

    def as_followers(self):
        """addresses the object to followers, if they are set"""
        if self.followers:
            self.to.add(self.followers)
        return self

    def as_unlisted(self):
        """makes the object unlisted, i.e. public in cc and followers in to"""
        if self.followers:
            self.to.add(self.followers)
        self.cc.add("https://www.w3.org/ns/activitystreams#Public")
        return self

    def now(self):
        """Sets published to now in isoformat"""
        self.published = now_isoformat()
        return self

    def build(self):
        """Returns the resulting object as a dictionary"""
        result = {
            "@context": "https://www.w3.org/ns/activitystreams",
            "type": self.type,
        }

        extra_fields = {
            "attributedTo": self.attributed_to,
            "to": list(self.to),
            "cc": list(self.cc - self.to),
            "id": self.id,
            "inReplyTo": self.in_reply_to,
            "published": self.published,
            "source": self.source,
            "name": self.name,
            "url": self.url,
            "summary": self.summary,
            "content": self.content,
            "tag": self.tag,
            "attachment": self.attachment,
            "href": self.href,
            "width": self.width,
            "height": self.height,
            "icon": self.icon,
            "mediaType": self.media_type,
        }

        for key, value in extra_fields.items():
            if value:
                result[key] = value

        if "to" in result and len(result["to"]) == 0:
            del result["to"]
        if "cc" in result and len(result["cc"]) == 0:
            del result["cc"]

        return result

as_followers()

addresses the object to followers, if they are set

Source code in bovine/bovine/activitystreams/object_factory.py 
50
51
52
53
54
def as_followers(self):
    """addresses the object to followers, if they are set"""
    if self.followers:
        self.to.add(self.followers)
    return self

as_public()

makes the object public, i.e. public in to and followers in cc

Source code in bovine/bovine/activitystreams/object_factory.py 
43
44
45
46
47
48
def as_public(self):
    """makes the object public, i.e. public in to and followers in cc"""
    self.to.add("https://www.w3.org/ns/activitystreams#Public")
    if self.followers:
        self.cc.add(self.followers)
    return self

as_unlisted()

makes the object unlisted, i.e. public in cc and followers in to

Source code in bovine/bovine/activitystreams/object_factory.py 
56
57
58
59
60
61
def as_unlisted(self):
    """makes the object unlisted, i.e. public in cc and followers in to"""
    if self.followers:
        self.to.add(self.followers)
    self.cc.add("https://www.w3.org/ns/activitystreams#Public")
    return self

build()

Returns the resulting object as a dictionary

Source code in bovine/bovine/activitystreams/object_factory.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def build(self):
    """Returns the resulting object as a dictionary"""
    result = {
        "@context": "https://www.w3.org/ns/activitystreams",
        "type": self.type,
    }

    extra_fields = {
        "attributedTo": self.attributed_to,
        "to": list(self.to),
        "cc": list(self.cc - self.to),
        "id": self.id,
        "inReplyTo": self.in_reply_to,
        "published": self.published,
        "source": self.source,
        "name": self.name,
        "url": self.url,
        "summary": self.summary,
        "content": self.content,
        "tag": self.tag,
        "attachment": self.attachment,
        "href": self.href,
        "width": self.width,
        "height": self.height,
        "icon": self.icon,
        "mediaType": self.media_type,
    }

    for key, value in extra_fields.items():
        if value:
            result[key] = value

    if "to" in result and len(result["to"]) == 0:
        del result["to"]
    if "cc" in result and len(result["cc"]) == 0:
        del result["cc"]

    return result

now()

Sets published to now in isoformat

Source code in bovine/bovine/activitystreams/object_factory.py 
63
64
65
66
def now(self):
    """Sets published to now in isoformat"""
    self.published = now_isoformat()
    return self

ObjectFactory

ObjectFactory usually created through a BovineClient

Source code in bovine/bovine/activitystreams/object_factory.py 
108
109
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
136
137
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
164
165
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
class ObjectFactory:
    """ObjectFactory usually created through a BovineClient"""

    def __init__(self, actor_information=None, client=None):
        if client:
            self.client = client
            self.information = client.information
        elif actor_information:
            self.client = None
            self.information = actor_information
        else:
            raise TypeError(
                "You need to either specify actor_information or a BovineClient"
            )

    def note(self, **kwargs):
        """Creates a Note Object"""
        return Object(
            attributed_to=self.information["id"],
            type="Note",
            followers=self.information.get("followers"),
            **kwargs,
        )

    def reply(self, obj: dict, **kwargs):
        """Creates a reply for an object

        :param obj: Object being replied to"""
        cc = (set(obj.get("to", [])) | set(obj.get("cc", []))) - {
            self.information["id"]
        }
        return Object(
            attributed_to=self.information["id"],
            type=obj.get("type", "Note"),
            in_reply_to=obj.get("id"),
            followers=self.information.get("followers"),
            to={id_for_object(obj.get("attributedTo"))},
            cc=cc,
            **kwargs,
        )

    def article(self, **kwargs):
        """Creates an Article Object"""
        return Object(
            attributed_to=self.information["id"],
            type="Article",
            followers=self.information.get("followers"),
            **kwargs,
        )

    def event(self, **kwargs):
        """Creates an Event Object"""
        return Object(
            attributed_to=self.information["id"],
            type="Event",
            followers=self.information.get("followers"),
            **kwargs,
        )

    async def mention_for_actor_uri(self, actor_to_mention):
        """Creates a mention object for another actor. Requires client to be set.

        :param actor_to_mention: The URI of the actor to mention"""

        if not self.client:
            raise TypeError("client needs to be set at construction")

        remote_actor = await self.client.proxy(actor_to_mention)

        return Object(
            type="Mention",
            href=actor_to_mention,
            name=fediverse_handle_from_actor(remote_actor),
        )

    async def reply_with_mention(self, obj: dict, **kwargs):
        """Creates a reply for an object, mentioning the author of the original
        post. This is necessary for compatibility with Mastodon

        :param obj: Object being replied to"""
        cc = (set(obj.get("to", [])) | set(obj.get("cc", []))) - {
            self.information["id"]
        }
        original_author = id_for_object(obj.get("attributedTo"))
        mention = (await self.mention_for_actor_uri(original_author)).build()
        return Object(
            attributed_to=self.information["id"],
            type=obj.get("type", "Note"),
            in_reply_to=obj.get("id"),
            followers=self.information.get("followers"),
            to={original_author},
            cc=cc,
            tag=[mention],
            **kwargs,
        )

article(**kwargs)

Creates an Article Object

Source code in bovine/bovine/activitystreams/object_factory.py 
149
150
151
152
153
154
155
156
def article(self, **kwargs):
    """Creates an Article Object"""
    return Object(
        attributed_to=self.information["id"],
        type="Article",
        followers=self.information.get("followers"),
        **kwargs,
    )

event(**kwargs)

Creates an Event Object

Source code in bovine/bovine/activitystreams/object_factory.py 
158
159
160
161
162
163
164
165
def event(self, **kwargs):
    """Creates an Event Object"""
    return Object(
        attributed_to=self.information["id"],
        type="Event",
        followers=self.information.get("followers"),
        **kwargs,
    )

mention_for_actor_uri(actor_to_mention) async

Creates a mention object for another actor. Requires client to be set.

Parameters:

Name Type Description Default
actor_to_mention

The URI of the actor to mention

 required
Source code in bovine/bovine/activitystreams/object_factory.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
async def mention_for_actor_uri(self, actor_to_mention):
    """Creates a mention object for another actor. Requires client to be set.

    :param actor_to_mention: The URI of the actor to mention"""

    if not self.client:
        raise TypeError("client needs to be set at construction")

    remote_actor = await self.client.proxy(actor_to_mention)

    return Object(
        type="Mention",
        href=actor_to_mention,
        name=fediverse_handle_from_actor(remote_actor),
    )

note(**kwargs)

Creates a Note Object

Source code in bovine/bovine/activitystreams/object_factory.py 
123
124
125
126
127
128
129
130
def note(self, **kwargs):
    """Creates a Note Object"""
    return Object(
        attributed_to=self.information["id"],
        type="Note",
        followers=self.information.get("followers"),
        **kwargs,
    )

reply(obj, **kwargs)

Creates a reply for an object

Parameters:

Name Type Description Default
obj dict

Object being replied to

 required
Source code in bovine/bovine/activitystreams/object_factory.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def reply(self, obj: dict, **kwargs):
    """Creates a reply for an object

    :param obj: Object being replied to"""
    cc = (set(obj.get("to", [])) | set(obj.get("cc", []))) - {
        self.information["id"]
    }
    return Object(
        attributed_to=self.information["id"],
        type=obj.get("type", "Note"),
        in_reply_to=obj.get("id"),
        followers=self.information.get("followers"),
        to={id_for_object(obj.get("attributedTo"))},
        cc=cc,
        **kwargs,
    )

reply_with_mention(obj, **kwargs) async

Creates a reply for an object, mentioning the author of the original post. This is necessary for compatibility with Mastodon

Parameters:

Name Type Description Default
obj dict

Object being replied to

 required
Source code in bovine/bovine/activitystreams/object_factory.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
async def reply_with_mention(self, obj: dict, **kwargs):
    """Creates a reply for an object, mentioning the author of the original
    post. This is necessary for compatibility with Mastodon

    :param obj: Object being replied to"""
    cc = (set(obj.get("to", [])) | set(obj.get("cc", []))) - {
        self.information["id"]
    }
    original_author = id_for_object(obj.get("attributedTo"))
    mention = (await self.mention_for_actor_uri(original_author)).build()
    return Object(
        attributed_to=self.information["id"],
        type=obj.get("type", "Note"),
        in_reply_to=obj.get("id"),
        followers=self.information.get("followers"),
        to={original_author},
        cc=cc,
        tag=[mention],
        **kwargs,
    )