Skip to content

bovine.activitystreams.collection_helper

CollectionHelper

Provides the ability to iterate over a Collection or OrderedCollection

Usage:

async for item in CollectionHelper(collection_id, bovine_client):
    do_something(item)

By setting resolve=True, items are always returned as a dictionary. Otherwise, they are returned as a string or dictionary depending on how the data is provided by the remote server.

Source code in bovine/bovine/activitystreams/collection_helper.py 
 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
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
class CollectionHelper:
    """Provides the ability to iterate over a Collection or OrderedCollection

    Usage:

    ```python
    async for item in CollectionHelper(collection_id, bovine_client):
        do_something(item)
    ```

    By setting resolve=True, items are always returned as a dictionary.
    Otherwise, they are returned as a string or dictionary depending on how
    the data is provided by the remote server."""

    def __init__(self, collection_id: str, actor, resolve=False):
        self.actor = actor
        self.collection_id = collection_id
        self.collection = collection_id
        self.items: list = []
        self.resolve = resolve
        self.fetched_first = False

    def update_items(self):
        if not isinstance(self.collection, dict):
            return False
        if "orderedItems" in self.collection:
            self.items = [*self.collection["orderedItems"]]
            return len(self.items) > 0
        if "items" in self.collection:
            self.items = [*self.collection["items"]]
            return len(self.items) > 0
        return False

    async def resolve_item(self, item):
        if not self.resolve or isinstance(item, dict):
            return item
        return await self.actor.proxy(item)

    def __aiter__(self):
        return self

    async def __anext__(self):
        if len(self.items) > 0:
            return await self.resolve_item(self.items.pop(0))
        if isinstance(self.collection, str):
            self.collection = await self.actor.proxy(self.collection)
            if self.update_items():
                return await self.resolve_item(self.items.pop(0))
        if not self.fetched_first and "first" in self.collection:
            self.fetched_first = True
            self.collection = self.collection["first"]
            if self.update_items():
                return await self.resolve_item(self.items.pop(0))
            else:
                return await self.__anext__()
        if "next" in self.collection:
            self.collection = self.collection["next"]
            if self.update_items():
                return await self.resolve_item(self.items.pop(0))
            else:
                return await self.__anext__()

        self.items = []
        self.collection = self.collection_id
        raise StopAsyncIteration

    async def as_collection(self) -> dict:
        """Returns an ActivityStreams collection containing the items represented
        by the collection_id. This is useful for collections spread over multiple
        pages"""

        items = [x async for x in self]
        if isinstance(self.collection, dict):
            collection_id = self.collection.get("part_of")
            if not collection_id:
                collection_id = self.collection["id"]
        else:
            collection_id = self.collection
        return Collection(id=collection_id, items=items).build()

as_collection() async

Returns an ActivityStreams collection containing the items represented by the collection_id. This is useful for collections spread over multiple pages

Source code in bovine/bovine/activitystreams/collection_helper.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
async def as_collection(self) -> dict:
    """Returns an ActivityStreams collection containing the items represented
    by the collection_id. This is useful for collections spread over multiple
    pages"""

    items = [x async for x in self]
    if isinstance(self.collection, dict):
        collection_id = self.collection.get("part_of")
        if not collection_id:
            collection_id = self.collection["id"]
    else:
        collection_id = self.collection
    return Collection(id=collection_id, items=items).build()