Skip to content

bovine_process.common.interactions

interaction_handlers = {None: dict(Announce=announce_handler, Create=reply_handler, Delete=delete_reply_handler, Dislike=like_handler, Like=like_handler, Undo=undo_handler), 'http://litepub.social/ns#EmojiReact': like_handler} module-attribute

The handlers being called for interactions

InteractionActor

Bases: ABC

Abstract class to illustrate what is needed to process interactions

Source code in bovine_process/bovine_process/common/interactions.py 
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
class InteractionActor(ABC):
    """Abstract class to illustrate what is needed to process interactions"""

    @abstractmethod
    async def retrieve(self, object_id: str, only_own: bool = False) -> dict | None:
        """Retrieves an object

        :param object_id: id of the object
        :param only_own: if true requires object to be by the actor"""
        pass

    @abstractmethod
    async def remove_references(self, object_id: str) -> None:
        """Removes references to the object. Used in case an object is
        deleted and it is unknown which object it replies to"""
        pass

    @abstractmethod
    async def add_to_interaction(
        self, interaction_type: str, object_id: str, interaction: str
    ) -> None:
        """Adds to interaction

        Example object for `interaction_type = "likes"`

        ```json
        {
            "type": "Like",
            "id": interaction,
            "object": object_id
        }
        ```

        :param interaction_type: One of `likes`, `shares, `replies`
        :param object_id: id of the object being interacted with
        :param interaction: The id of the interacting object
        """
        pass

    @abstractmethod
    async def remove_from_interaction(
        self, interaction_type: str, object_id: str, interaction: str
    ) -> None:
        """Removes from interaction

        :param interaction_type: One of `likes`, `shares, `replies`
        :param object_id: id of the object being interacted with
        :param interaction: The id of the interacting object"""

        pass

add_to_interaction(interaction_type, object_id, interaction) abstractmethod async

Adds to interaction

Example object for interaction_type = "likes"

{
    "type": "Like",
    "id": interaction,
    "object": object_id
}

Parameters:

Name Type Description Default
interaction_type str

One of likes, shares,replies`

 required
object_id str

id of the object being interacted with

 required
interaction str

The id of the interacting object

 required
Source code in bovine_process/bovine_process/common/interactions.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@abstractmethod
async def add_to_interaction(
    self, interaction_type: str, object_id: str, interaction: str
) -> None:
    """Adds to interaction

    Example object for `interaction_type = "likes"`

    ```json
    {
        "type": "Like",
        "id": interaction,
        "object": object_id
    }
    ```

    :param interaction_type: One of `likes`, `shares, `replies`
    :param object_id: id of the object being interacted with
    :param interaction: The id of the interacting object
    """
    pass

remove_from_interaction(interaction_type, object_id, interaction) abstractmethod async

Removes from interaction

Parameters:

Name Type Description Default
interaction_type str

One of likes, shares,replies`

 required
object_id str

id of the object being interacted with

 required
interaction str

The id of the interacting object

 required
Source code in bovine_process/bovine_process/common/interactions.py 
54
55
56
57
58
59
60
61
62
63
64
@abstractmethod
async def remove_from_interaction(
    self, interaction_type: str, object_id: str, interaction: str
) -> None:
    """Removes from interaction

    :param interaction_type: One of `likes`, `shares, `replies`
    :param object_id: id of the object being interacted with
    :param interaction: The id of the interacting object"""

    pass

remove_references(object_id) abstractmethod async

Removes references to the object. Used in case an object is deleted and it is unknown which object it replies to

Source code in bovine_process/bovine_process/common/interactions.py 
26
27
28
29
30
@abstractmethod
async def remove_references(self, object_id: str) -> None:
    """Removes references to the object. Used in case an object is
    deleted and it is unknown which object it replies to"""
    pass

retrieve(object_id, only_own=False) abstractmethod async

Retrieves an object

Parameters:

Name Type Description Default
object_id str

id of the object

 required
only_own bool

if true requires object to be by the actor

 False
Source code in bovine_process/bovine_process/common/interactions.py 
18
19
20
21
22
23
24
@abstractmethod
async def retrieve(self, object_id: str, only_own: bool = False) -> dict | None:
    """Retrieves an object

    :param object_id: id of the object
    :param only_own: if true requires object to be by the actor"""
    pass

announce_handler(item, actor) async

Adds object to the shares collection of the announced object, if

  • object being announced is owner by the receiving actor
Source code in bovine_process/bovine_process/common/interactions.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def announce_handler(
    item: ProcessingItem, actor: InteractionActor
) -> ProcessingItem:
    """Adds object to the shares collection of the announced object, if

    * object being announced is owner by the receiving actor"""

    obj = await own_object(item, actor)
    if obj:
        obj_id = id_for_object(obj)
        if obj_id is None:
            return item
        logger.info("Announce Handler %s", obj_id)
        await actor.add_to_interaction("shares", obj_id, item.data.get("id"))

    return item

delete_reply_handler(item, actor) async

If a reply is deleted, removes it from the replies collection

Source code in bovine_process/bovine_process/common/interactions.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def delete_reply_handler(
    item: ProcessingItem, actor: InteractionActor
) -> ProcessingItem:
    """If a reply is deleted, removes it from the replies collection"""

    remote = item.data.get("object")
    if not remote:
        return item

    if isinstance(remote, dict) and remote.get("type") == "Person":
        return item

    await actor.remove_references(remote)

    return item

like_handler(item, actor) async

Adds object to the likes collection of the liked object, if

  • object being liked is owner by the receiving actor
Source code in bovine_process/bovine_process/common/interactions.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def like_handler(item: ProcessingItem, actor: InteractionActor) -> ProcessingItem:
    """Adds object to the likes collection of the liked object, if

    * object being liked is owner by the receiving actor"""

    obj = await own_object(item, actor)
    if obj:
        obj_id = id_for_object(obj)
        if obj_id is None:
            return item
        logger.info("Like Handler %s", obj_id)
        await actor.add_to_interaction("likes", obj_id, item.data.get("id"))

    return item

reply_handler(item, actor) async

Adds object to the replies collection. Object being replied to is determined from inReplyTo. Reply is added if the object belongs to the receiving actor.

Source code in bovine_process/bovine_process/common/interactions.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
async def reply_handler(
    item: ProcessingItem, actor: InteractionActor
) -> ProcessingItem:
    """Adds object to the replies collection. Object being replied to
    is determined from `inReplyTo`. Reply is added if the object
    belongs to the receiving actor."""
    create = Activity(item.data, domain=item.submitter_domain)
    remote = await create.object_for_create(actor.retrieve)

    if not remote:
        return item

    if not remote.in_reply_to:
        return item

    obj = await actor.retrieve(remote.in_reply_to, only_own=True)

    if obj:
        obj_id = id_for_object(obj)
        if obj_id is None:
            return item
        logger.info("Reply Handler %s", obj_id)
        await actor.add_to_interaction("replies", obj_id, remote.identifier)

    return item

undo_handler(item, actor) async

For an Undo of a Like, Dislike, Announce , they are removed from the appropriate collection.

Source code in bovine_process/bovine_process/common/interactions.py 
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
async def undo_handler(item: ProcessingItem, actor: InteractionActor) -> ProcessingItem:
    """For an Undo of a Like, Dislike, Announce , they are removed from
    the appropriate collection."""

    object_to_undo = id_for_object(item.data.get("object"))
    if object_to_undo is None:
        return item

    obj = await actor.retrieve(object_to_undo)
    if obj is None:
        return item

    remote_actor = id_for_object(item.data.get("actor"))
    if obj.get("actor") != remote_actor:
        logger.error("Mismatching actor in undo from %s", remote_actor)
        return item

    obj_type = obj.get("type")
    if obj_type in ["Like", "Dislike", "http://litepub.social/ns#EmojiReact"]:
        logger.info("Undo Handler for Like of %s", obj.get("id"))
        await actor.remove_from_interaction("likes", obj.get("object"), obj.get("id"))
    elif obj_type == "Announce":
        logger.info("Undo Handler for Like of %s", obj.get("id"))
        await actor.remove_from_interaction("shares", obj.get("object"), obj.get("id"))

    return item