Skip to content

roboherd

Main roboherd module. Should contain all necessary elements to build a robocow

Activity = Annotated[dict, Depends(get_activity)] module-attribute

The activity parsed by muck_out

ActivityFactory = Annotated[BovineActivityFactory, Depends(get_activity_factory)] module-attribute

The activity factory of type bovine.activitystreams.activity_factory.ActivityFactory

EmbeddedObject = Annotated[dict, Depends(get_embedded_object)] module-attribute

The embedded object in the activity as parsed by muck_out

MarkdownPoster = Annotated[MarkdownPublisher, Depends(get_markdown_poster)] module-attribute

A function that takes a markdown string and posts it as the content of a note

ObjectFactory = Annotated[BovineObjectFactory, Depends(get_object_factory)] module-attribute

The object factory of type bovine.activitystreams.object_factory.ObjectFactory

ParsedData = Annotated[dict, Depends(get_parsed)] module-attribute

The parsed data as transformed by muck_out

PublishActivity = Annotated[Publisher, Depends(construct_publish_activity)] module-attribute

Allows one to publish an activity as the actor. Assumes cattle_grid has the extension simple_object_storage or equivalent

PublishObject = Annotated[Publisher, Depends(construct_publish_object)] module-attribute

Allows one to publish an object as the actor. Assumes cattle_grid has the extension simple_object_storage or equivalent

RawData = Annotated[dict, Depends(get_raw)] module-attribute

The raw data as received by cattle_grid

RoboCow dataclass

RoboCow(information: roboherd.cow.types.Information, auto_follow: bool = True, skip_profile_update: bool = False, internals: roboherd.cow.RoboCowInternals = )

Parameters:

Name Type Description Default
information Information

Information about the cow

 required
auto_follow bool

Whether to automatically accept follow requests

 True
skip_profile_update bool

When set to True the profile is not updated automatically. Useful when managing a cow from multiple scripts.

 False
internals RoboCowInternals

Internal data for the cow

 <dynamic>
Source code in roboherd/cow/__init__.py 
 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
@dataclass
class RoboCow:
    information: Information = field(
        metadata=dict(description="Information about the cow")
    )

    auto_follow: bool = field(
        default=True,
        metadata=dict(
            description="""Whether to automatically accept follow requests"""
        ),
    )

    skip_profile_update: bool = field(
        default=False,
        metadata=dict(
            description="When set to True the profile is not updated automatically. Useful when managing a cow from multiple scripts."
        ),
    )

    internals: RoboCowInternals = field(
        default_factory=RoboCowInternals,
        metadata=dict(description="Internal data for the cow"),
    )

    @staticmethod
    def create(**kwargs):
        """Creates a new cow. We note that
        `RoboCow.create(name="my name")` is equivalent
        to `RoboCow(information=Information(name="my name"))`.
        """
        return RoboCow(information=Information(**kwargs))

    def action(self, action: str = "*", activity_type: str = "*"):
        """Adds a handler for an event. Use "*" as a wildcard.

        Usage:

        ```python
        cow = Robocow(information=Information(handle="example"))

        @cow.action(action="outgoing", activity_type="Follow")
        async def handle_outgoing_follow(data):
            ...
        ```
        """

        config = HandlerConfiguration(
            action=action,
            activity_type=activity_type,
        )

        def inner(func):
            config.func = func
            self.internals.handlers.add_handler(config, func)
            self.internals.handler_configuration.append(config)
            return func

        return inner

    def cron(self, crontab):
        def inner(func):
            self.internals.cron_entries.append(CronEntry(crontab, func))

            return func

        return inner

    def incoming(self, func):
        """Adds a handler for an incoming message. Usage:

        ```python
        cow = Robocow("example")

        @cow.incoming
        async def handle_incoming(data):
            ...
        ```
        """
        config = HandlerConfiguration(
            action="incoming",
            activity_type="*",
        )
        self.internals.handlers.add_handler(config, func)
        return func

    def incoming_create(self, func):
        """Adds a handler for an incoming activity if the
        activity is of type_create

        ```python
        cow = Robocow("example")

        @cow.incoming_create
        async def handle_incoming(data):
            ...
        ```
        """
        config = HandlerConfiguration(
            action="incoming", activity_type="Create", func=func
        )
        self.internals.handler_configuration.append(config)
        self.internals.handlers.add_handler(config, func)
        return func

    def startup(self, func):
        """Adds a startup routine to be run when the cow is started."""

        self.internals.startup_routine = func

    async def run_startup(self, connection: Almabtrieb):
        """Runs when the cow is birthed"""

        if self.internals.profile is None:
            if not self.internals.actor_id:
                raise ValueError("Actor ID is not set")
            result = await connection.fetch(
                self.internals.actor_id, self.internals.actor_id
            )
            if not result.data:
                raise ValueError("Could not retrieve profile")
            self.internals.profile = result.data

        if self.internals.cron_entries:
            frequency = ", ".join(
                get_description(entry.crontab) for entry in self.internals.cron_entries
            )
            self.information.frequency = frequency

        if not self.skip_profile_update:
            await self._run_profile_update(connection)

        if self.internals.startup_routine:
            await inject(self.internals.startup_routine)(
                cow=self,  # type:ignore
                connection=connection,  # type:ignore
                actor_id=self.internals.actor_id,  # type:ignore
            )  # type:ignore

    async def _run_profile_update(self, connection: Almabtrieb):
        if self.internals.profile is None:
            raise ValueError("Profile is not set")

        update = determine_profile_update(self.information, self.internals.profile)

        if update:
            logger.info("Updating profile for %s", self.information.handle)

            await connection.trigger("update_actor", update)

action(action='*', activity_type='*')

Adds a handler for an event. Use “*” as a wildcard.

Usage:

cow = Robocow(information=Information(handle="example"))

@cow.action(action="outgoing", activity_type="Follow")
async def handle_outgoing_follow(data):
    ...
Source code in roboherd/cow/__init__.py 
 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
def action(self, action: str = "*", activity_type: str = "*"):
    """Adds a handler for an event. Use "*" as a wildcard.

    Usage:

    ```python
    cow = Robocow(information=Information(handle="example"))

    @cow.action(action="outgoing", activity_type="Follow")
    async def handle_outgoing_follow(data):
        ...
    ```
    """

    config = HandlerConfiguration(
        action=action,
        activity_type=activity_type,
    )

    def inner(func):
        config.func = func
        self.internals.handlers.add_handler(config, func)
        self.internals.handler_configuration.append(config)
        return func

    return inner

create(**kwargs) staticmethod

Creates a new cow. We note that RoboCow.create(name="my name") is equivalent to RoboCow(information=Information(name="my name")).

Source code in roboherd/cow/__init__.py 
91
92
93
94
95
96
97
@staticmethod
def create(**kwargs):
    """Creates a new cow. We note that
    `RoboCow.create(name="my name")` is equivalent
    to `RoboCow(information=Information(name="my name"))`.
    """
    return RoboCow(information=Information(**kwargs))

incoming(func)

Adds a handler for an incoming message. Usage:

cow = Robocow("example")

@cow.incoming
async def handle_incoming(data):
    ...
Source code in roboherd/cow/__init__.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def incoming(self, func):
    """Adds a handler for an incoming message. Usage:

    ```python
    cow = Robocow("example")

    @cow.incoming
    async def handle_incoming(data):
        ...
    ```
    """
    config = HandlerConfiguration(
        action="incoming",
        activity_type="*",
    )
    self.internals.handlers.add_handler(config, func)
    return func

incoming_create(func)

Adds a handler for an incoming activity if the activity is of type_create

cow = Robocow("example")

@cow.incoming_create
async def handle_incoming(data):
    ...
Source code in roboherd/cow/__init__.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def incoming_create(self, func):
    """Adds a handler for an incoming activity if the
    activity is of type_create

    ```python
    cow = Robocow("example")

    @cow.incoming_create
    async def handle_incoming(data):
        ...
    ```
    """
    config = HandlerConfiguration(
        action="incoming", activity_type="Create", func=func
    )
    self.internals.handler_configuration.append(config)
    self.internals.handlers.add_handler(config, func)
    return func

run_startup(connection) async

Runs when the cow is birthed

Source code in roboherd/cow/__init__.py 
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
async def run_startup(self, connection: Almabtrieb):
    """Runs when the cow is birthed"""

    if self.internals.profile is None:
        if not self.internals.actor_id:
            raise ValueError("Actor ID is not set")
        result = await connection.fetch(
            self.internals.actor_id, self.internals.actor_id
        )
        if not result.data:
            raise ValueError("Could not retrieve profile")
        self.internals.profile = result.data

    if self.internals.cron_entries:
        frequency = ", ".join(
            get_description(entry.crontab) for entry in self.internals.cron_entries
        )
        self.information.frequency = frequency

    if not self.skip_profile_update:
        await self._run_profile_update(connection)

    if self.internals.startup_routine:
        await inject(self.internals.startup_routine)(
            cow=self,  # type:ignore
            connection=connection,  # type:ignore
            actor_id=self.internals.actor_id,  # type:ignore
        )  # type:ignore

startup(func)

Adds a startup routine to be run when the cow is started.

Source code in roboherd/cow/__init__.py 
171
172
173
174
def startup(self, func):
    """Adds a startup routine to be run when the cow is started."""

    self.internals.startup_routine = func