Skip to content

bovine.activitystreams.activity_factory

Activity dataclass

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

Source code in bovine/bovine/activitystreams/activity_factory.py 
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
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
@dataclass
class Activity:
    """A dataclass representing an `ActivityStreams Activity
    <https://www.w3.org/TR/activitystreams-vocabulary/#activity-types>`_"""

    type: str
    actor: 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

    target: Optional[str] = None
    object: Optional[str] = None

    def as_public(self):
        """makes the activity 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 activity to followers, if they are set"""
        if self.followers:
            self.to.add(self.followers)
        return self

    def as_unlisted(self):
        """makes the activity 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 build(self) -> dict:
        """converts the activity into a dict, that can be serialized to JSON"""
        result = {
            "@context": "https://www.w3.org/ns/activitystreams",
            "type": self.type,
            "actor": self.actor,
            "to": list(self.to),
            "cc": list(self.cc - self.to),
        }

        extra_fields = {
            "id": self.id,
            "published": self.published,
            "name": self.name,
            "summary": self.summary,
            "content": self.content,
            "target": self.target,
            "object": self.object,
        }

        if result["to"] is not None and len(result["to"]) == 0:
            del result["to"]
        if result["cc"] is not None and len(result["cc"]) == 0:
            del result["cc"]

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

        return result

as_followers()

addresses the activity to followers, if they are set

Source code in bovine/bovine/activitystreams/activity_factory.py 
39
40
41
42
43
def as_followers(self):
    """addresses the activity to followers, if they are set"""
    if self.followers:
        self.to.add(self.followers)
    return self

as_public()

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

Source code in bovine/bovine/activitystreams/activity_factory.py 
32
33
34
35
36
37
def as_public(self):
    """makes the activity 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 activity unlisted, i.e. public in cc and followers in to

Source code in bovine/bovine/activitystreams/activity_factory.py 
45
46
47
48
49
50
def as_unlisted(self):
    """makes the activity 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()

converts the activity into a dict, that can be serialized to JSON

Source code in bovine/bovine/activitystreams/activity_factory.py 
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
def build(self) -> dict:
    """converts the activity into a dict, that can be serialized to JSON"""
    result = {
        "@context": "https://www.w3.org/ns/activitystreams",
        "type": self.type,
        "actor": self.actor,
        "to": list(self.to),
        "cc": list(self.cc - self.to),
    }

    extra_fields = {
        "id": self.id,
        "published": self.published,
        "name": self.name,
        "summary": self.summary,
        "content": self.content,
        "target": self.target,
        "object": self.object,
    }

    if result["to"] is not None and len(result["to"]) == 0:
        del result["to"]
    if result["cc"] is not None and len(result["cc"]) == 0:
        del result["cc"]

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

    return result

ActivityFactory

Basic factory for Activity objects.

Usally created by a BovineClient

Source code in bovine/bovine/activitystreams/activity_factory.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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
class ActivityFactory:
    """Basic factory for Activity objects.

    Usally created by a BovineClient"""

    def __init__(self, actor_information):
        self.information = actor_information

    def create(self, obj, **kwargs):
        """Activity of type Create from Object"""
        return Activity(
            type="Create",
            actor=obj.get("attributedTo", self.information["id"]),
            cc=set(obj.get("cc", [])),
            to=set(obj.get("to", [])),
            object=obj,
            **kwargs,
        )

    def update(self, obj, **kwargs):
        """Activity of type Update from Object"""
        return Activity(
            type="Update",
            actor=obj.get("attributedTo", self.information["id"]),
            cc=set(obj.get("cc", [])),
            to=set(obj.get("to", [])),
            object=obj,
            **kwargs,
        )

    def like(self, target, **kwargs):
        """Like for target"""
        return Activity(
            type="Like", actor=self.information["id"], object=target, **kwargs
        )

    def delete(self, target, **kwargs):
        """Delete for target"""
        return Activity(
            type="Delete", actor=self.information["id"], object=target, **kwargs
        )

    def accept(self, obj, **kwargs):
        """Accept for object"""
        if isinstance(obj, str):
            return Activity(
                type="Accept", actor=self.information["id"], object=obj, **kwargs
            )
        return Activity(
            type="Accept",
            actor=self.information["id"],
            object=obj,
            to={id_for_object(obj.get("actor"))},
            **kwargs,
        )

    def reject(self, obj, **kwargs):
        """Reject for object"""
        return Activity(
            type="Reject",
            actor=self.information["id"],
            object=obj,
            to={id_for_object(obj.get("actor"))},
            **kwargs,
        )

    def announce(self, obj, **kwargs):
        """Announce for object"""
        return Activity(
            type="Announce",
            actor=self.information["id"],
            object=obj,
            followers=self.information.get("followers", []),
            **kwargs,
        )

    def follow(self, obj, **kwargs):
        """Follow for object"""

        return Activity(
            type="Follow",
            actor=self.information["id"],
            object=obj,
            to={id_for_object(obj)},
            **kwargs,
        )

    def undo(self, activity, **kwargs):
        """Undo for activity"""

        return Activity(
            type="Undo",
            actor=self.information["id"],
            object=activity,
            to={id_for_object(x) for x in activity.get("to")},
            **kwargs,
        )

accept(obj, **kwargs)

Accept for object

Source code in bovine/bovine/activitystreams/activity_factory.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
def accept(self, obj, **kwargs):
    """Accept for object"""
    if isinstance(obj, str):
        return Activity(
            type="Accept", actor=self.information["id"], object=obj, **kwargs
        )
    return Activity(
        type="Accept",
        actor=self.information["id"],
        object=obj,
        to={id_for_object(obj.get("actor"))},
        **kwargs,
    )

announce(obj, **kwargs)

Announce for object

Source code in bovine/bovine/activitystreams/activity_factory.py 
150
151
152
153
154
155
156
157
158
def announce(self, obj, **kwargs):
    """Announce for object"""
    return Activity(
        type="Announce",
        actor=self.information["id"],
        object=obj,
        followers=self.information.get("followers", []),
        **kwargs,
    )

create(obj, **kwargs)

Activity of type Create from Object

Source code in bovine/bovine/activitystreams/activity_factory.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def create(self, obj, **kwargs):
    """Activity of type Create from Object"""
    return Activity(
        type="Create",
        actor=obj.get("attributedTo", self.information["id"]),
        cc=set(obj.get("cc", [])),
        to=set(obj.get("to", [])),
        object=obj,
        **kwargs,
    )

delete(target, **kwargs)

Delete for target

Source code in bovine/bovine/activitystreams/activity_factory.py 
120
121
122
123
124
def delete(self, target, **kwargs):
    """Delete for target"""
    return Activity(
        type="Delete", actor=self.information["id"], object=target, **kwargs
    )

follow(obj, **kwargs)

Follow for object

Source code in bovine/bovine/activitystreams/activity_factory.py 
160
161
162
163
164
165
166
167
168
169
def follow(self, obj, **kwargs):
    """Follow for object"""

    return Activity(
        type="Follow",
        actor=self.information["id"],
        object=obj,
        to={id_for_object(obj)},
        **kwargs,
    )

like(target, **kwargs)

Like for target

Source code in bovine/bovine/activitystreams/activity_factory.py 
114
115
116
117
118
def like(self, target, **kwargs):
    """Like for target"""
    return Activity(
        type="Like", actor=self.information["id"], object=target, **kwargs
    )

reject(obj, **kwargs)

Reject for object

Source code in bovine/bovine/activitystreams/activity_factory.py 
140
141
142
143
144
145
146
147
148
def reject(self, obj, **kwargs):
    """Reject for object"""
    return Activity(
        type="Reject",
        actor=self.information["id"],
        object=obj,
        to={id_for_object(obj.get("actor"))},
        **kwargs,
    )

undo(activity, **kwargs)

Undo for activity

Source code in bovine/bovine/activitystreams/activity_factory.py 
171
172
173
174
175
176
177
178
179
180
def undo(self, activity, **kwargs):
    """Undo for activity"""

    return Activity(
        type="Undo",
        actor=self.information["id"],
        object=activity,
        to={id_for_object(x) for x in activity.get("to")},
        **kwargs,
    )

update(obj, **kwargs)

Activity of type Update from Object

Source code in bovine/bovine/activitystreams/activity_factory.py 
103
104
105
106
107
108
109
110
111
112
def update(self, obj, **kwargs):
    """Activity of type Update from Object"""
    return Activity(
        type="Update",
        actor=obj.get("attributedTo", self.information["id"]),
        cc=set(obj.get("cc", [])),
        to=set(obj.get("to", [])),
        object=obj,
        **kwargs,
    )