Skip to content

bovine_process.utils

ByActivityType

Simplifies handling activity by type, usage:

by_type = ByActivityType(One=do_one, Two=do_two})

by_type({"type": one}, arg1, arg2) # calls do_one({"type": one}, arg1, arg2)
Source code in bovine_process/bovine_process/utils/__init__.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
class ByActivityType:
    """Simplifies handling activity by type, usage:

    ```python
    by_type = ByActivityType(One=do_one, Two=do_two})

    by_type({"type": one}, arg1, arg2) # calls do_one({"type": one}, arg1, arg2)
    ```
    """

    def __init__(self, **kwargs):
        self.actions = kwargs

    async def __call__(self, item, *args):
        try:
            item_type = item.data.get("type")
            if item_type is None:
                return item
            if isinstance(item_type, str):
                if item_type in self.actions:
                    return await self.actions[item_type](item, *args)
                else:
                    return item
            elif isinstance(item_type, list):
                for single_type in item_type:
                    item = await self.actions[single_type](item, *args)
                return item
            else:
                return item

        except Exception as ex:
            logger.error("Something went wrong with %s during procession", repr(ex))
            logger.exception(ex)
            if item:
                logger.error(item.data)

ProcessorList

Creates a list of processors, usage:

processors = ProcessorList(my_processor,
    ByActivityType(Follow=my_follow_processor))
await processors(item, args)
Source code in bovine_process/bovine_process/utils/__init__.py 
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
class ProcessorList:
    """Creates a list of processors, usage:

    ```python
    processors = ProcessorList(my_processor,
        ByActivityType(Follow=my_follow_processor))
    await processors(item, args)
    ```"""

    def __init__(self, *processors):
        self.processors = list(processors)

    async def __call__(self, item, *arguments):
        working = item

        try:
            for processor in self.processors:
                working = await processor(working, *arguments)
                if not working:
                    return

            return working
        except Exception as ex:
            logger.exception(ex)
            raise ex