roboherd.cow

CronEntry dataclass

A cron entry

Parameters:

Name	Type	Description	Default
crontab	str	The cron expression	required
func	Callable	The function to be called	required

Source code in roboherd/cow/__init__.py

@dataclass
class CronEntry:
    """A cron entry"""

    crontab: str = field(metadata=dict(description="""The cron expression"""))

    func: Callable = field(metadata=dict(description="""The function to be called"""))

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

@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

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

@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

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:
        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

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

    self.internals.startup_routine = func

RoboCowInternals dataclass

Internal data for the cow

Parameters:

Name	Type	Description	Default
profile	dict | None	The profile of the cow, aka as the actor object in ActivityPub	None
actor_id	str | None	Actor Id of the cow; loaded automatically	None
handlers	Handlers	Handlers for incoming and outgoing messages, added through annotations	<roboherd.cow.handlers.Handlers object at 0x7f7f3dc39810>
handler_configuration	List[HandlerConfiguration]	Handler configurations, added through annotations	<dynamic>
cron_entries	List[CronEntry]	Cron entries, created through annotations	<dynamic>
startup_routine	Callable | None		None
base_url	str | None		None

Source code in roboherd/cow/__init__.py

@dataclass
class RoboCowInternals:
    """Internal data for the cow"""

    profile: dict | None = field(
        default=None,
        metadata=dict(
            description="""The profile of the cow, aka as the actor object in ActivityPub"""
        ),
    )

    actor_id: str | None = field(
        default=None,
        metadata=dict(description="""Actor Id of the cow; loaded automatically"""),
    )

    handlers: Handlers = field(
        default_factory=Handlers,
        metadata=dict(
            description="""Handlers for incoming and outgoing messages, added through annotations"""
        ),
    )
    handler_configuration: List[HandlerConfiguration] = field(
        default_factory=list,
        metadata=dict(
            description="""Handler configurations, added through annotations"""
        ),
    )

    cron_entries: List[CronEntry] = field(
        default_factory=list,
        metadata=dict(description="""Cron entries, created through annotations"""),
    )

    startup_routine: Callable | None = None

    base_url: str | None = field(default=None)

const

default_icon = {'mediaType': 'image/png', 'type': 'Image', 'url': 'https://dev.bovine.social/assets/bull-horns.png'} module-attribute

The default icon to be used

handlers

HandlerConfiguration dataclass

HandlerConfiguration(action: str, activity_type: str, func: Optional[Callable] = None)

Parameters:

Name	Type	Description	Default
action	str		required
activity_type	str		required
func	Callable | None		None

Source code in roboherd/cow/handlers.py

@dataclass
class HandlerConfiguration:
    action: str
    activity_type: str
    func: Callable | None = None

profile

determine_profile_update(information, profile)

Returns the update for the profile

Source code in roboherd/cow/profile.py

def determine_profile_update(information: Information, profile: dict) -> dict | None:
    """Returns the update for the profile"""

    update = {"actor": profile.get("id")}

    if profile_part_needs_update(information, profile):
        update["profile"] = {
            "type": information.type,
            "name": information.name,
            "summary": information.description,
            "icon": information.icon,
        }

    actions = determine_actions(information, profile)

    if actions:
        update["actions"] = actions

    if len(update) == 1:
        return None

    return update

types

Information

Bases: BaseModel

Information about the cow

Parameters:

Name	Type	Description	Default
type	str	ActivityPub type of the actor.	'Service'
handle	str | None	Used as the handle in acct:handle@domain.example	None
name	str | None	The display name of the cow	None
description	str | None	The description of the cow, used as summary of the actor	None
icon	dict	The profile image	{'mediaType': 'image/png', 'type': 'Image', 'url': 'https://dev.bovine.social/assets/bull-horns.png'}
frequency	str | None	Frequency of posting. Is set automatically if cron expressions are used.	None
meta_information	MetaInformation	Meta information about the cow, such as the source repository	MetaInformation(source=None, author=None)

Source code in roboherd/cow/types.py

class Information(BaseModel):
    """Information about the cow"""

    type: str = Field(
        default="Service",
        examples=["Service"],
        description="ActivityPub type of the actor.",
    )

    handle: str | None = Field(
        default=None,
        examples=["moocow"],
        description="Used as the handle in `acct:handle@domain.example`",
    )

    name: str | None = Field(
        default=None,
        examples=["The mooing cow 🐮"],
        description="The display name of the cow",
    )

    description: str | None = Field(
        default=None,
        examples=[
            "I'm a cow that moos.",
            """<p>An example bot to illustrate Roboherd</p><p>For more information on RoboHerd, see <a href="https://codeberg.org/bovine/roboherd">its repository</a>.</p>""",
        ],
        description="The description of the cow, used as summary of the actor",
    )

    icon: dict = Field(
        default=default_icon,
        description="The profile image",
    )

    frequency: str | None = Field(
        default=None,
        examples=["daily"],
        description="Frequency of posting. Is set automatically if cron expressions are used.",
    )

    meta_information: MetaInformation = Field(
        default=MetaInformation(),
        description="Meta information about the cow, such as the source repository",
    )

MetaInformation

Bases: BaseModel

Meta Information about the bot. This includes information such as the author and the source repository

Parameters:

Name	Type	Description	Default
source	str | None	The source repository	None
author	str | None	The author, often a Fediverse handle	None

Source code in roboherd/cow/types.py

class MetaInformation(BaseModel):
    """Meta Information about the bot. This includes
    information such as the author and the source repository"""

    source: str | None = Field(
        default=None,
        examples=["https://forge.example/repo"],
        description="The source repository",
    )

    author: str | None = Field(
        default=None,
        examples=["acct:author@domain.example"],
        description="The author, often a Fediverse handle",
    )

util

HandlerInformation dataclass

HandlerInformation(func: Callable[[Dict], Awaitable[NoneType]])

Parameters:

Name	Type	Description	Default
func	Callable[list, Awaitable[None]]		required

Source code in roboherd/cow/util.py

@dataclass
class HandlerInformation:
    func: Callable[[Dict], Awaitable[None]]