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, name: str = ‘’, config: dict = , state: dict = , 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
name	str		''
config	dict	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
state	dict	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
internals	RoboCowInternals	Internal data for the cow	<dynamic>

Source code in roboherd/cow/__init__.py

@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."
        ),
    )

    name: str = field(default="")
    config: dict = field(default_factory=dict)
    state: dict = field(default_factory=dict)

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

    @staticmethod
    def create(**kwargs):
        """Creates a new cow, by creating a new [Information][roboherd.cow.types.Information].

        ```python
        >>> RoboCow.create(name="my cow")
        RoboCow(information=Information(type='Service', handle=None, name='my cow', ...

        ```

        The parameter `description_md` is transformed from markdown to html and then
        assigned to `description`.

        ```python
        >> cow = RoboCow.create(description_md="__bold__")
        >> cow.information.description
        '<p><strong>bold</strong></p>'

        ```
        """

        if "description_md" in kwargs:
            import markdown  # type: ignore

            kwargs["description"] = markdown.markdown(kwargs["description_md"])
            del kwargs["description_md"]

        information = Information.model_validate(kwargs)
        return RoboCow(information=information)

    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:
            self.state = await inject(self.internals.startup_routine)(
                cow=self,  # type:ignore
                connection=connection,  # type:ignore
                actor_id=self.internals.actor_id,  # type:ignore
                config=self.config,
                state=self.state,
            )  # type:ignore

            print(self.state)

    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

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, by creating a new Information.

>>> RoboCow.create(name="my cow")
RoboCow(information=Information(type='Service', handle=None, name='my cow', ...

The parameter description_md is transformed from markdown to html and then assigned to description.

>> cow = RoboCow.create(description_md="__bold__")
>> cow.information.description
'<p><strong>bold</strong></p>'

Source code in roboherd/cow/__init__.py

@staticmethod
def create(**kwargs):
    """Creates a new cow, by creating a new [Information][roboherd.cow.types.Information].

    ```python
    >>> RoboCow.create(name="my cow")
    RoboCow(information=Information(type='Service', handle=None, name='my cow', ...

    ```

    The parameter `description_md` is transformed from markdown to html and then
    assigned to `description`.

    ```python
    >> cow = RoboCow.create(description_md="__bold__")
    >> cow.information.description
    '<p><strong>bold</strong></p>'

    ```
    """

    if "description_md" in kwargs:
        import markdown  # type: ignore

        kwargs["description"] = markdown.markdown(kwargs["description_md"])
        del kwargs["description_md"]

    information = Information.model_validate(kwargs)
    return RoboCow(information=information)

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

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

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

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:
        self.state = await inject(self.internals.startup_routine)(
            cow=self,  # type:ignore
            connection=connection,  # type:ignore
            actor_id=self.internals.actor_id,  # type:ignore
            config=self.config,
            state=self.state,
        )  # type:ignore

        print(self.state)

startup(func)

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

Source code in roboherd/cow/__init__.py

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

    self.internals.startup_routine = func