Skip to content

bovine_store.actor.bovine_store_actor

BovineStoreActor

Bases: BovineActor

Represents an actor controlled by bovine and stored in the database. This class should be used to perform actions as this actor. In particular, instances of this class are passed to bovine_process when performing side effects.

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
 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
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
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
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
class BovineStoreActor(BovineActor):
    """Represents an actor controlled by bovine and stored in the database.
    This class should be used to perform actions as this actor. In particular,
    instances of this class are passed to bovine_process when performing
    side effects."""

    def __init__(
        self, actor_object: Actor, endpoints: List[str] = [], bovine_name=None, **config
    ):
        self.actor_object: Actor = actor_object
        self.bovine_name = bovine_name

        self.domain = domain_from_host(actor_object.id)
        self.endpoints = endpoints

        super().__init__(**config)

    async def retrieve_own_object(self, object_id: str) -> dict | None:
        """Deprecated in favor of `retrieve(..., only_own=True)`"""
        warnings.warn(
            "Deprecated use retrieve(...,only_own=True) instead, will be remove in bovine 0.6",
            DeprecationWarning,
        )
        return await self._retrieve_own_object(object_id)

    async def _retrieve_own_object(self, object_id: str) -> dict | None:
        result = await StoredJsonObject.get_or_none(id=object_id)
        if not result:
            return None
        if result.owner != self.actor_object.id:
            return None
        return result.content

    async def retrieve(
        self,
        object_id: str,
        include: List[str] = [],
        skip_fetch: bool = False,
        only_own: bool = False,
    ) -> dict | None:
        """Retrieves the object with identified by object_id. The logic is
        as follows:

        * If object is in database and either local or remote and last updated
            in the last day, the object is returned from the database
        * Otherwise the object is fetched (as the actor) unless skip_fetch is set
        * Finally if include lists properties to be resolved, these objects
            are obtained using the above logic, and then added to the Object

        :param object_id: The object id to fetch
        :param include: list of keys, who should be resolved against the database
        :param skip_fetch: set to true to not attempt to fetch remote object
        :param only_own: Only returns if object is owned by actor

        :return: The resulting object as a dictionary if successful otherwise `None`.
        """
        if isinstance(object_id, dict):
            object_id = object_id.get("id")
        if not object_id:
            return None
        if only_own:
            return await self._retrieve_own_object(object_id)
        if len(object_id) > 255:
            logger.warning("Got too long object_id %s", object_id)
            return None

        data = await self._retrieve_object_from_database(
            object_id, skip_fetch=skip_fetch
        )

        if data is None:
            return data

        if len(include) == 0:
            return data

        items = await asyncio.gather(
            *[
                self._retrieve_object_from_database(data[key])
                for key in include
                if key in data
            ]
        )

        return combine_items(data, items)

    async def _retrieve_object_from_database(
        self, object_id, skip_fetch: bool = False
    ) -> Optional[dict]:
        if isinstance(object_id, dict):
            return object_id
        try:
            result = await StoredJsonObject.get_or_none(id=object_id)

            if skip_fetch:
                if result:
                    return result.content
                return None

            if not BovineStoreActor._is_stale(result):
                return result.content

            logger.info("Object %s is being refetched", object_id)

            obj = await self.get(object_id)
            if "attributedTo" in obj:
                owner = obj["attributedTo"]
            elif "actor" in obj:
                owner = obj["actor"]
            else:
                owner = object_id

            logger.info("Storing %s", obj["id"])

            await update_remote_object(
                owner,
                obj,
                visible_to=[self.actor_object.id],
                object_type=ObjectType.REMOTE,
            )

            return obj
        except Exception as ex:
            logger.info("Failed to retrieve %s with %s", object_id, ex)
            for log_line in traceback.format_exc().splitlines():
                logger.info(log_line)

            return None

    async def store(self, item: dict, as_public: bool = False, visible_to=[]):
        """Stores object in database"""
        return await store_remote_object(
            self.actor_id,
            item,
            as_public=as_public,
            visible_to=visible_to,
            object_type=ObjectType.LOCAL,
        )

    async def update(self, item: dict):
        """Updates object in database"""
        await update_remote_object(self.actor_object.id, item)

    async def retrieve_for(self, retriever, object_id, include=[]):
        return await retrieve_remote_object(retriever, object_id, include=include)

    async def store_for(self, owner: str, item: dict, as_public=False, visible_to=[]):
        """Stores a remote object

        :param owner: The actor who has submitted the object
        :param item: The object to store"""

        return await store_remote_object(
            owner, item, as_public=as_public, visible_to=visible_to, domain=self.domain
        )

    async def update_for(self, owner, item):
        return await update_remote_object(owner, item)

    async def add_to_outbox(self, object_id):
        return await add_to_collection(self.actor_object.outbox, object_id)

    async def add_to_inbox(self, object_id):
        return await add_to_collection(self.actor_object.inbox, object_id)

    async def add_to_followers(self, object_id):
        """Add object_id to followers collection"""
        return await add_to_collection(self.actor_object.followers, object_id)

    async def remove_from_followers(self, object_id):
        """Remove object_id from followers collection"""
        return await remove_from_collection(self.actor_object.followers, object_id)

    async def add_to_following(self, object_id):
        """Add object_id to following collection"""
        return await add_to_collection(self.actor_object.following, object_id)

    async def remove_from_following(self, object_id):
        """Remove object_id from following collection"""
        return await remove_from_collection(self.actor_object.following, object_id)

    async def add_to_interaction(
        self, interaction: str, object_id: str, remote_id: str
    ):
        """Adds to an interaction collection

        :param interaction:
            The interaction either replies, shares, or likes
        :param object_id:
            id of the object being interacted with
        :param remote_id:
            id of the interaction, e.g. of the Like or Dislike
        """
        return await add_to_collection(f"{object_id}/{interaction}", remote_id)

    async def remove_from_interaction(
        self, interaction: str, object_id: str, remote_id: str
    ):
        """Removes an interaction from the corresponding collection

        :param interaction:
            The interaction either replies, shares, or likes
        :param object_id:
            id of the object being interacted with
        :param remote_id:
            id of the interaction, e.g. of the Like or Dislike
        """
        obj = await CollectionItem.get_or_none(
            part_of=f"{object_id}/{interaction}", object_id=remote_id
        )
        if obj:
            await obj.delete()

    async def remove_references(self, remote_id: str):
        """Remove remote_id from replies collections"""
        references = await CollectionItem.filter(object_id=remote_id).all()
        for reply in references:
            if reply.part_of.endswith("replies"):
                await reply.delete()

    async def resolve_endpoints(self, endpoints):
        tasks = [collection_all(self.actor_id, endpoint) for endpoint in endpoints]
        result = await asyncio.gather(*tasks)
        result = set(sum(result, []))

        logger.info("Resolved %s to %s", ", ".join(endpoints), ", ".join(result))
        return result

    def generate_new_object_id(self) -> str:
        """Creates a new object id

        This contains a hard coded path; probably should replace with something better.
        """
        return urljoin(self.actor_object.id, "/objects/" + str(uuid.uuid4()))

    async def update_profile(self, data: dict):
        """Updates the Actor profile"""

        database_actor = await DbBovineActor.get(bovine_name=self.bovine_name)
        database_actor.properties = data
        await database_actor.save()

    @staticmethod
    def _is_stale(result):
        if result is None:
            return True
        if result.object_type == ObjectType.LOCAL:
            return False

        updated = result.updated
        updated = updated.replace(tzinfo=timezone.utc)

        return updated < (datetime.now(tz=timezone.utc) - timedelta(days=1))

    @staticmethod
    async def from_database(stored_actor, session):
        """Creates BovineStoreAction from database object

        :param stored_actor:
            The BovineActor object from the database
        :param session:
            An aiohttp.ClientSession"""
        mapped_endpoints = {x.endpoint_type: x for x in stored_actor.endpoints}
        account_url = mapped_endpoints[EndpointType.ACTOR].name

        def get_server_key(stored_actor):
            for iter_keypair in stored_actor.keypairs:
                if iter_keypair.name == "serverKey":
                    return iter_keypair
            return stored_actor.keypairs[0]

        keypair = get_server_key(stored_actor)
        public_key_url = f"{account_url}#{keypair.name}"

        endpoints = [
            x.name
            for x in stored_actor.endpoints
            if x.endpoint_type in [EndpointType.FOLLOWERS, EndpointType.COLLECTION]
        ]

        actor_object = Actor(
            id=account_url,
            preferred_username=stored_actor.handle_name,
            inbox=mapped_endpoints[EndpointType.INBOX].name,
            outbox=mapped_endpoints[EndpointType.OUTBOX].name,
            event_source=mapped_endpoints[EndpointType.EVENT_SOURCE].name,
            followers=mapped_endpoints[EndpointType.FOLLOWERS].name,
            following=mapped_endpoints[EndpointType.FOLLOWING].name,
            proxy_url=mapped_endpoints[EndpointType.PROXY_URL].name,
            public_key=keypair.public_key,
            public_key_name=keypair.name,
            properties=stored_actor.properties,
        )

        if "type" in stored_actor.properties:
            actor_object.type = stored_actor.properties["type"]
        if "name" in stored_actor.properties:
            actor_object.name = stored_actor.properties["name"]
        if "summary" in stored_actor.properties:
            actor_object.summary = stored_actor.properties["summary"]
        if "icon" in stored_actor.properties:
            actor_object.icon = stored_actor.properties["icon"]
        if "url" in stored_actor.properties:
            actor_object.url = stored_actor.properties["url"]

        actor = BovineStoreActor(
            actor_object,
            endpoints=endpoints,
            bovine_name=stored_actor.bovine_name,
            actor_id=account_url,
            public_key_url=public_key_url,
            secret=keypair.private_key,
        )
        await actor.init(session=session)

        return actor

add_to_followers(object_id) async

Add object_id to followers collection

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
203
204
205
async def add_to_followers(self, object_id):
    """Add object_id to followers collection"""
    return await add_to_collection(self.actor_object.followers, object_id)

add_to_following(object_id) async

Add object_id to following collection

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
211
212
213
async def add_to_following(self, object_id):
    """Add object_id to following collection"""
    return await add_to_collection(self.actor_object.following, object_id)

add_to_interaction(interaction, object_id, remote_id) async

Adds to an interaction collection

Parameters:

Name Type Description Default
interaction str

The interaction either replies, shares, or likes

 required
object_id str

id of the object being interacted with

 required
remote_id str

id of the interaction, e.g. of the Like or Dislike

 required
Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
async def add_to_interaction(
    self, interaction: str, object_id: str, remote_id: str
):
    """Adds to an interaction collection

    :param interaction:
        The interaction either replies, shares, or likes
    :param object_id:
        id of the object being interacted with
    :param remote_id:
        id of the interaction, e.g. of the Like or Dislike
    """
    return await add_to_collection(f"{object_id}/{interaction}", remote_id)

from_database(stored_actor, session) async staticmethod

Creates BovineStoreAction from database object

Parameters:

Name Type Description Default
stored_actor

The BovineActor object from the database

 required
session

An aiohttp.ClientSession

 required
Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
@staticmethod
async def from_database(stored_actor, session):
    """Creates BovineStoreAction from database object

    :param stored_actor:
        The BovineActor object from the database
    :param session:
        An aiohttp.ClientSession"""
    mapped_endpoints = {x.endpoint_type: x for x in stored_actor.endpoints}
    account_url = mapped_endpoints[EndpointType.ACTOR].name

    def get_server_key(stored_actor):
        for iter_keypair in stored_actor.keypairs:
            if iter_keypair.name == "serverKey":
                return iter_keypair
        return stored_actor.keypairs[0]

    keypair = get_server_key(stored_actor)
    public_key_url = f"{account_url}#{keypair.name}"

    endpoints = [
        x.name
        for x in stored_actor.endpoints
        if x.endpoint_type in [EndpointType.FOLLOWERS, EndpointType.COLLECTION]
    ]

    actor_object = Actor(
        id=account_url,
        preferred_username=stored_actor.handle_name,
        inbox=mapped_endpoints[EndpointType.INBOX].name,
        outbox=mapped_endpoints[EndpointType.OUTBOX].name,
        event_source=mapped_endpoints[EndpointType.EVENT_SOURCE].name,
        followers=mapped_endpoints[EndpointType.FOLLOWERS].name,
        following=mapped_endpoints[EndpointType.FOLLOWING].name,
        proxy_url=mapped_endpoints[EndpointType.PROXY_URL].name,
        public_key=keypair.public_key,
        public_key_name=keypair.name,
        properties=stored_actor.properties,
    )

    if "type" in stored_actor.properties:
        actor_object.type = stored_actor.properties["type"]
    if "name" in stored_actor.properties:
        actor_object.name = stored_actor.properties["name"]
    if "summary" in stored_actor.properties:
        actor_object.summary = stored_actor.properties["summary"]
    if "icon" in stored_actor.properties:
        actor_object.icon = stored_actor.properties["icon"]
    if "url" in stored_actor.properties:
        actor_object.url = stored_actor.properties["url"]

    actor = BovineStoreActor(
        actor_object,
        endpoints=endpoints,
        bovine_name=stored_actor.bovine_name,
        actor_id=account_url,
        public_key_url=public_key_url,
        secret=keypair.private_key,
    )
    await actor.init(session=session)

    return actor

generate_new_object_id()

Creates a new object id

This contains a hard coded path; probably should replace with something better.

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
266
267
268
269
270
271
def generate_new_object_id(self) -> str:
    """Creates a new object id

    This contains a hard coded path; probably should replace with something better.
    """
    return urljoin(self.actor_object.id, "/objects/" + str(uuid.uuid4()))

remove_from_followers(object_id) async

Remove object_id from followers collection

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
207
208
209
async def remove_from_followers(self, object_id):
    """Remove object_id from followers collection"""
    return await remove_from_collection(self.actor_object.followers, object_id)

remove_from_following(object_id) async

Remove object_id from following collection

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
215
216
217
async def remove_from_following(self, object_id):
    """Remove object_id from following collection"""
    return await remove_from_collection(self.actor_object.following, object_id)

remove_from_interaction(interaction, object_id, remote_id) async

Removes an interaction from the corresponding collection

Parameters:

Name Type Description Default
interaction str

The interaction either replies, shares, or likes

 required
object_id str

id of the object being interacted with

 required
remote_id str

id of the interaction, e.g. of the Like or Dislike

 required
Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
async def remove_from_interaction(
    self, interaction: str, object_id: str, remote_id: str
):
    """Removes an interaction from the corresponding collection

    :param interaction:
        The interaction either replies, shares, or likes
    :param object_id:
        id of the object being interacted with
    :param remote_id:
        id of the interaction, e.g. of the Like or Dislike
    """
    obj = await CollectionItem.get_or_none(
        part_of=f"{object_id}/{interaction}", object_id=remote_id
    )
    if obj:
        await obj.delete()

remove_references(remote_id) async

Remove remote_id from replies collections

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
251
252
253
254
255
256
async def remove_references(self, remote_id: str):
    """Remove remote_id from replies collections"""
    references = await CollectionItem.filter(object_id=remote_id).all()
    for reply in references:
        if reply.part_of.endswith("replies"):
            await reply.delete()

retrieve(object_id, include=[], skip_fetch=False, only_own=False) async

Retrieves the object with identified by object_id. The logic is as follows:

  • If object is in database and either local or remote and last updated in the last day, the object is returned from the database
  • Otherwise the object is fetched (as the actor) unless skip_fetch is set
  • Finally if include lists properties to be resolved, these objects are obtained using the above logic, and then added to the Object

Parameters:

Name Type Description Default
object_id str

The object id to fetch

 required
include List[str]

list of keys, who should be resolved against the database

 []
skip_fetch bool

set to true to not attempt to fetch remote object

 False
only_own bool

Only returns if object is owned by actor

 False

Returns:

Type Description
dict | None

The resulting object as a dictionary if successful otherwise None.
Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def retrieve(
    self,
    object_id: str,
    include: List[str] = [],
    skip_fetch: bool = False,
    only_own: bool = False,
) -> dict | None:
    """Retrieves the object with identified by object_id. The logic is
    as follows:

    * If object is in database and either local or remote and last updated
        in the last day, the object is returned from the database
    * Otherwise the object is fetched (as the actor) unless skip_fetch is set
    * Finally if include lists properties to be resolved, these objects
        are obtained using the above logic, and then added to the Object

    :param object_id: The object id to fetch
    :param include: list of keys, who should be resolved against the database
    :param skip_fetch: set to true to not attempt to fetch remote object
    :param only_own: Only returns if object is owned by actor

    :return: The resulting object as a dictionary if successful otherwise `None`.
    """
    if isinstance(object_id, dict):
        object_id = object_id.get("id")
    if not object_id:
        return None
    if only_own:
        return await self._retrieve_own_object(object_id)
    if len(object_id) > 255:
        logger.warning("Got too long object_id %s", object_id)
        return None

    data = await self._retrieve_object_from_database(
        object_id, skip_fetch=skip_fetch
    )

    if data is None:
        return data

    if len(include) == 0:
        return data

    items = await asyncio.gather(
        *[
            self._retrieve_object_from_database(data[key])
            for key in include
            if key in data
        ]
    )

    return combine_items(data, items)

retrieve_own_object(object_id) async

Deprecated in favor of retrieve(..., only_own=True)

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
55
56
57
58
59
60
61
async def retrieve_own_object(self, object_id: str) -> dict | None:
    """Deprecated in favor of `retrieve(..., only_own=True)`"""
    warnings.warn(
        "Deprecated use retrieve(...,only_own=True) instead, will be remove in bovine 0.6",
        DeprecationWarning,
    )
    return await self._retrieve_own_object(object_id)

store(item, as_public=False, visible_to=[]) async

Stores object in database

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
167
168
169
170
171
172
173
174
175
async def store(self, item: dict, as_public: bool = False, visible_to=[]):
    """Stores object in database"""
    return await store_remote_object(
        self.actor_id,
        item,
        as_public=as_public,
        visible_to=visible_to,
        object_type=ObjectType.LOCAL,
    )

store_for(owner, item, as_public=False, visible_to=[]) async

Stores a remote object

Parameters:

Name Type Description Default
owner str

The actor who has submitted the object

 required
item dict

The object to store

 required
Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
184
185
186
187
188
189
190
191
192
async def store_for(self, owner: str, item: dict, as_public=False, visible_to=[]):
    """Stores a remote object

    :param owner: The actor who has submitted the object
    :param item: The object to store"""

    return await store_remote_object(
        owner, item, as_public=as_public, visible_to=visible_to, domain=self.domain
    )

update(item) async

Updates object in database

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
177
178
179
async def update(self, item: dict):
    """Updates object in database"""
    await update_remote_object(self.actor_object.id, item)

update_profile(data) async

Updates the Actor profile

Source code in bovine_store/bovine_store/actor/bovine_store_actor.py 
273
274
275
276
277
278
async def update_profile(self, data: dict):
    """Updates the Actor profile"""

    database_actor = await DbBovineActor.get(bovine_name=self.bovine_name)
    database_actor.properties = data
    await database_actor.save()