cattle_grid.model

cattle_grid.model

ActivityMessage

Bases: WithActor

Message that contains an Activity. Activity is used as the name for the ‘data object’ being exchanged, as is common in the Fediverse

Parameters:

Name	Type	Description	Default
actor	str	actor_id of the actor that received the message	required
data	Dict[str, Any]	The activity	required

Source code in cattle_grid/model/__init__.py

class ActivityMessage(WithActor):
    """
    Message that contains an Activity. Activity is used as the name for the 'data object' being exchanged, as is common in the Fediverse
    """

    model_config = ConfigDict(
        extra="forbid",
    )

    data: Dict[str, Any] = Field(description="The activity")

FetchMessage

Bases: WithActor

Used to request an ActivityPub object to be retrieved

Parameters:

Name	Type	Description	Default
actor	str	actor_id of the actor that received the message	required
uri	str	URI of the object being retrieved	required

Source code in cattle_grid/model/__init__.py

class FetchMessage(WithActor):
    """
    Used to request an ActivityPub object to be retrieved
    """

    model_config = ConfigDict(
        extra="forbid",
    )

    uri: str = Field(description="URI of the object being retrieved")

SharedInboxMessage

Bases: BaseModel

Message that contains an Activity. In difference to the ActivityMessage this message does not have an actor, and thus its recipients will be determined by cattle_grid.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	Activity	required

Source code in cattle_grid/model/__init__.py

class SharedInboxMessage(BaseModel):
    """
    Message that contains an Activity. In difference to the ActivityMessage this message does not have an actor, and thus its recipients will be determined by cattle_grid.
    """

    model_config = ConfigDict(
        extra="forbid",
    )
    data: Dict[str, Any] = Field(description="Activity")

account

Objects used for the account exchange

ActorInformation

Bases: BaseModel

Information about an actor

Parameters:

Name	Type	Description	Default
id	str	The URI corresponding to the actor	required
name	str	Internal name of the actor	required

Source code in cattle_grid/model/account.py

class ActorInformation(BaseModel):
    """Information about an actor"""

    id: str = Field(
        description="The URI corresponding to the actor",
        examples=["http://host.example/actor/1"],
    )
    name: str = Field(examples=["Alice"], description="Internal name of the actor")

CreateActorRequest

Bases: BaseModel

Request to create an actor

Parameters:

Name	Type	Description	Default
base_url	str	Base url for the actor, the actor URI will be of the form {base_url}/actor/{id}	required
name	str | None	Internal name of the actor	None
preferred_username	str | None	Add a preferred username. This name will be used in acct:username@domain and supplied to webfinger. Here domain is determine from baseUrl.	None
profile	Dict[str, Any]	New profile object for the actor.	{}
automatically_accept_followers	bool | None	Enables setting actors to automatically accept follow requests	None

Source code in cattle_grid/model/account.py

class CreateActorRequest(BaseModel):
    """Request to create an actor"""

    base_url: str = Field(
        examples=["http://host.example"],
        serialization_alias="baseUrl",
        description="""Base url for the actor, the actor URI will be of the form `{base_url}/actor/{id}`""",
    )
    name: str | None = Field(
        default=None, examples=["alice"], description="""Internal name of the actor"""
    )

    preferred_username: str | None = Field(
        default=None,
        examples=["alice", "bob"],
        description="""
    Add a preferred username. This name will be used in acct:username@domain and supplied to webfinger. Here domain is determine from baseUrl.
    """,
        serialization_alias="preferredUsername",
    )
    profile: Dict[str, Any] = Field(
        default={},
        examples=[{"summary": "A new actor"}],
        description="""
    New profile object for the actor.
    """,
    )
    automatically_accept_followers: bool | None = Field(
        default=None,
        examples=[True, False, None],
        description="""
    Enables setting actors to automatically accept follow requests
    """,
        serialization_alias="automaticallyAcceptFollowers",
    )

ErrorMessage

Bases: BaseModel

Message to send an exception

Parameters:

Name	Type	Description	Default
message	list[str]	The exception	required
routing_key	str	The routing key of the message that caused the error	required

Source code in cattle_grid/model/account.py

class ErrorMessage(BaseModel):
    """Message to send an exception"""

    model_config = ConfigDict(extra="allow")

    message: list[str] = Field(description="The exception")

    routing_key: str = Field(
        examples=["send.alice.trigger.method"],
        description="""The routing key of the message that caused the error""",
    )

EventInformation

Bases: WithActor, WithTransformedData

Send on outgoing or incoming events

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The data that was exchanged. We note that this data was processed by the transformers.	required
actor	str	actor_id of the actor that received the message	required
event_type	EventType	Type of event incoming means that the contained data was not generated on the actors behalf. outgoing means that the data is being send out by the actor.	required

Source code in cattle_grid/model/account.py

class EventInformation(WithActor, WithTransformedData):
    """Send on outgoing or incoming events"""

    event_type: EventType = Field(
        examples=[EventType.incoming, EventType.outgoing],
        description="""Type of event

incoming means that the contained data was not generated on the actors behalf.

outgoing means that the data is being send out by the actor.""",
    )

EventType

Bases: StrEnum

Types of events

>>> EventType.incoming.value
'incoming'

Source code in cattle_grid/model/account.py

class EventType(StrEnum):
    """Types of events

    ```pycon
    >>> EventType.incoming.value
    'incoming'

    ```
    """

    incoming = auto()
    outgoing = auto()
    error = auto()

FetchMessage

Bases: WithActor

Message to fetch an object from the Fediverse

Parameters:

Name	Type	Description	Default
actor	str	actor_id of the actor that received the message	required
uri	str	The resource to fetch	required

Source code in cattle_grid/model/account.py

class FetchMessage(WithActor):
    """Message to fetch an object from the Fediverse"""

    uri: str = Field(
        examples=["http://remote.example/object/1"],
        description="""The resource to fetch""",
    )

FetchResponse

Bases: WithActor

Result of a a fetch request

Parameters:

Name	Type	Description	Default
actor	str	actor_id of the actor that received the message	required
uri	str	The resource that was requested	required
data	dict | None	The data returned for the object	required

Source code in cattle_grid/model/account.py

class FetchResponse(WithActor):
    """Result of a a fetch request"""

    uri: str = Field(
        examples=["http://remote.example/object/1"],
        description="""The resource that was requested""",
    )

    data: dict | None = Field(description="""The data returned for the object""")

InformationResponse

Bases: BaseModel

Response for the information request

Parameters:

Name	Type	Description	Default
account_name	str	Name of the account	required
actors	List[ActorInformation]	Actors of the account on the server	required
base_urls	List[str]	The base urls of the server	required
backend	NameAndVersion	Name and version of the backend	required
protocol	NameAndVersion	Name and version of the protocol being used	required
method_information	List[MethodInformationModel]	Built-in mutable sequence. If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.	<dynamic>

Source code in cattle_grid/model/account.py

class InformationResponse(BaseModel):
    """Response for the information request"""

    account_name: str = Field(
        examples=["alice"],
        serialization_alias="accountName",
        description="Name of the account",
    )

    actors: List[ActorInformation] = Field(
        examples=[
            [
                ActorInformation(id="http://host.example/actor/1", name="Alice"),
                ActorInformation(id="http://host.example/actor/2", name="Bob"),
            ]
        ],
        description="Actors of the account on the server",
    )

    base_urls: List[str] = Field(
        examples=[["http://host.example"]],
        serialization_alias="baseUrls",
        description="""The base urls of the server""",
    )

    backend: NameAndVersion = Field(
        examples=[NameAndVersion(name="cattle_grid", version="3.1.4")],
        description="""Name and version of the backend""",
    )

    protocol: NameAndVersion = Field(
        examples=[NameAndVersion(name="CattleDrive", version="3.1.4")],
        description="""Name and version of the protocol being used""",
    )

    method_information: List[MethodInformationModel] = Field(
        default_factory=list,
        examples=[
            [
                MethodInformationModel(
                    routing_key="send_message",
                    module="cattle_grid",
                    description="Send a message as the actor",
                )
            ]
        ],
        serialization_alias="methodInformation",
    )

NameAndVersion

Bases: BaseModel

Name and version information

Parameters:

Name	Type	Description	Default
name	str	Name of the server or protocol	required
version	str	Version of the server or protocol	required

Source code in cattle_grid/model/account.py

class NameAndVersion(BaseModel):
    """Name and version information"""

    name: str = Field(
        examples=["cattle_grid", "CattleDrive"],
        description="""Name of the server or protocol""",
    )

    version: str = Field(
        examples=["3.1.4"],
        description="""Version of the server or protocol""",
    )

TriggerMessage

Bases: WithActor

Message to trigger something on the ActivityExchange

Parameters:

Name	Type	Description	Default
actor	str	actor_id of the actor that received the message	required

Source code in cattle_grid/model/account.py

class TriggerMessage(WithActor):
    """Message to trigger something on the ActivityExchange"""

    model_config = ConfigDict(extra="allow")

common

WithActor

Bases: BaseModel

Parameters:

Name	Type	Description	Default
actor	str	actor_id of the actor that received the message	required

Source code in cattle_grid/model/common.py

class WithActor(BaseModel):
    actor: str = Field(
        description="actor_id of the actor that received the message",
        examples=["http://host.example/actor"],
    )

WithTransformedData

Bases: BaseModel

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The data that was exchanged. We note that this data was processed by the transformers.	required

Source code in cattle_grid/model/common.py

class WithTransformedData(BaseModel):
    data: dict[str, Any] = Field(
        examples=[
            {
                "raw": {
                    "@context": "https://www.w3.org/ns/activitystreams",
                    "type": "Create",
                    "actor": "http://host.example/actor/1",
                }
            }
        ],
        description="""The data that was exchanged. We note that this data was processed by the transformers.""",
    )

exchange

DeleteActorMessage

Bases: WithActor

Allows one to delete the actor object

Parameters:

Name	Type	Description	Default
actor	str	actor_id of the actor that received the message	required

Source code in cattle_grid/model/exchange.py

class DeleteActorMessage(WithActor):
    """
    Allows one to delete the actor object
    """

    model_config = ConfigDict(
        extra="forbid",
    )

TransformedActivityMessage

Bases: WithActor, WithTransformedData

Transformed activity message

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The data that was exchanged. We note that this data was processed by the transformers.	required
actor	str	actor_id of the actor that received the message	required

Source code in cattle_grid/model/exchange.py

class TransformedActivityMessage(WithActor, WithTransformedData):
    """Transformed activity message"""

UpdateActorMessage

Bases: WithActor

Allows one to update the actor object

Parameters:

Name	Type	Description	Default
actor	str	actor_id of the actor that received the message	required
profile	Dict[str, Any] | None	New profile object for the actor. The fields.	None
autoFollow	bool | None	Enables setting actors to automatically accept follow requests	None
actions	List[Annotated[Union[UpdateIdentifierAction, UpdatePropertyValueAction, UpdateUrlAction, RenameActorAction], FieldInfo]]	Actions to be taken when updating the profile	<dynamic>

Source code in cattle_grid/model/exchange.py

class UpdateActorMessage(WithActor):
    """
    Allows one to update the actor object
    """

    # model_config = ConfigDict(
    #     extra="forbid",
    # )

    profile: Dict[str, Any] | None = Field(
        default=None,
        examples=[{"summary": "A new description of the actor"}],
        description="""
    New profile object for the actor. The fields.
    """,
    )
    autoFollow: bool | None = Field(
        default=None,
        examples=[True, False, None],
        description="""
    Enables setting actors to automatically accept follow requests
    """,
    )

    actions: List[
        Annotated[
            UpdateAction,
            Field(
                discriminator="action",
            ),
        ]
    ] = Field(
        default_factory=list,
        description="""Actions to be taken when updating the profile""",
    )

    @field_serializer("actions")
    def serialize_dt(self, actions: List[UpdateAction], _info):
        return [action.model_dump() for action in actions]

exchange_update_actor

RenameActorAction

Bases: BaseModel

Update the internal name of the actor

Parameters:

Name	Type	Description	Default
action	Literal[<UpdateActionType.rename: 'rename'>]		<UpdateActionType.rename: 'rename'>
name	str	The new name of the actor	required

Source code in cattle_grid/model/exchange_update_actor.py

class RenameActorAction(BaseModel):
    """Update the internal name of the actor"""

    action: Literal[UpdateActionType.rename] = Field(default=UpdateActionType.rename)

    name: str = Field(description="The new name of the actor")

UpdateActionType

Bases: StrEnum

Available actions for updating the actor

Source code in cattle_grid/model/exchange_update_actor.py

class UpdateActionType(StrEnum):
    """Available actions for updating the actor"""

    add_identifier = auto()
    """Adds a new identifier. The identifier is assumed to already exist."""
    create_identifier = auto()
    """Creates a new identifier. Must be on a domain controlled by cattle_grid and enabled in the account"""
    update_identifier = auto()
    """Updates an identifer"""
    remove_identifier = auto()
    """Removes an identifier"""

    rename = auto()
    """Updates the internal name of the actor"""

    update_property_value = auto()
    """Adds or updates a property value of the actor"""
    remove_property_value = auto()
    """Removes a property value"""

    add_url = auto()
    """Adds an url to the actor profile"""
    remove_url = auto()
    """Removes an url from the actor profile"""

add_identifier class-attribute instance-attribute

add_identifier = auto()

Adds a new identifier. The identifier is assumed to already exist.

add_url class-attribute instance-attribute

add_url = auto()

Adds an url to the actor profile

create_identifier class-attribute instance-attribute

create_identifier = auto()

Creates a new identifier. Must be on a domain controlled by cattle_grid and enabled in the account

remove_identifier class-attribute instance-attribute

remove_identifier = auto()

Removes an identifier

remove_property_value class-attribute instance-attribute

remove_property_value = auto()

Removes a property value

remove_url class-attribute instance-attribute

remove_url = auto()

Removes an url from the actor profile

rename class-attribute instance-attribute

rename = auto()

Updates the internal name of the actor

update_identifier class-attribute instance-attribute

update_identifier = auto()

Updates an identifer

update_property_value class-attribute instance-attribute

update_property_value = auto()

Adds or updates a property value of the actor

UpdateIdentifierAction

Bases: BaseModel

Used to update an identifier of the actor

Parameters:

Name	Type	Description	Default
action	Literal[<UpdateActionType.add_identifier: 'add_identifier'>, <UpdateActionType.create_identifier: 'create_identifier'>, <UpdateActionType.update_identifier: 'update_identifier'>]		required
identifier	str	The identifier	required
primary	bool	Set the identifier as the primary one, if the identifier corresponds to an acct-uri this will update the primary identifier	False

Source code in cattle_grid/model/exchange_update_actor.py

class UpdateIdentifierAction(BaseModel):
    """Used to update an identifier of the actor"""

    action: Literal[
        UpdateActionType.add_identifier,
        UpdateActionType.create_identifier,
        UpdateActionType.update_identifier,
    ]

    identifier: str = Field(
        description="The identifier", examples=["acct:alice@domain.example"]
    )
    primary: bool = Field(
        False,
        description="Set the identifier as the primary one, if the identifier corresponds to an acct-uri this will update the primary identifier",
    )

UpdatePropertyValueAction

Bases: BaseModel

Update a property value of the actor

Parameters:

Name	Type	Description	Default
action	Literal[<UpdateActionType.update_property_value: 'update_property_value'>, <UpdateActionType.remove_property_value: 'remove_property_value'>]		required
key	str	The key of the property value to be created, updated, or deleted	required
value	str | None	The value of the property value	None

Source code in cattle_grid/model/exchange_update_actor.py

class UpdatePropertyValueAction(BaseModel):
    """Update a property value of the actor"""

    action: Literal[
        UpdateActionType.update_property_value, UpdateActionType.remove_property_value
    ]

    key: str = Field(
        examples=["author"],
        description="The key of the property value to be created, updated, or deleted",
    )
    value: str | None = Field(
        None,
        examples=["Alice"],
        description="The value of the property value",
    )

UpdateUrlAction

Bases: BaseModel

Parameters:

Name	Type	Description	Default
action	Literal[<UpdateActionType.add_url: 'add_url'>, <UpdateActionType.remove_url: 'remove_url'>]		required
url	str	The url to add	required
media_type	str | None	The media type for the url	None
rel	str | None	The relation to use	None

Source code in cattle_grid/model/exchange_update_actor.py

class UpdateUrlAction(BaseModel):
    action: Literal[UpdateActionType.add_url, UpdateActionType.remove_url]

    url: str = Field(description="The url to add")
    media_type: str | None = Field(
        default=None, description="The media type for the url"
    )
    rel: str | None = Field(default=None, description="The relation to use")

extension

AddMethodInformationMessage

Bases: BaseModel

Message to add method information through the exchange

Parameters:

Name	Type	Description	Default
method_information	list[MethodInformationModel]	List of method information	required

Source code in cattle_grid/model/extension.py

class AddMethodInformationMessage(BaseModel):
    """Message to add method information through the exchange"""

    method_information: list[MethodInformationModel] = Field(
        description="""List of method information""",
    )

MethodInformationModel

Bases: BaseModel

cattle_grid allows to define methods on the exchange through extensions. This class contains a description of them

Parameters:

Name	Type	Description	Default
routing_key	str	Name of the method	required
module	str	Module the extension was imported from. This is cattle_grid for build-in methods	required
description	str | None	Description of the method	None
replies	bool	Indicates whether the method will reply on receive.NAME.response.trigger to the trigger event	False

Source code in cattle_grid/model/extension.py

class MethodInformationModel(BaseModel):
    """cattle_grid allows to define methods on the
    exchange through extensions. This class contains
    a description of them"""

    routing_key: str = Field(
        examples=["send_message"],
        description="""Name of the method""",
    )

    module: str = Field(
        examples=["cattle_grid"],
        description="""Module the extension was imported from. This is cattle_grid for build-in methods""",
    )

    description: str | None = Field(
        default=None,
        examples=["Send a message as the actor"],
        description="""Description of the method""",
    )

    replies: bool = Field(
        default=False,
        description="Indicates whether the method will reply on `receive.NAME.response.trigger` to the trigger event",
    )

lookup

LookupMethod module-attribute

LookupMethod = Callable[[Lookup], Awaitable[Lookup]]

Alias for the Lookup Method

Lookup

Bases: BaseModel

Lookup of something from the Fediverse

Parameters:

Name	Type	Description	Default
uri	str	The uri being looked up	required
actor	str	The actor performing the lookup	required
result	dict | None	The result of the lookup, None if no result yet, the result will be returned once the lookup is finished	None

Source code in cattle_grid/model/lookup.py

class Lookup(BaseModel):
    """
    Lookup of something from the Fediverse
    """

    uri: str = Field(
        examples=["http://actor.example", "acct:user@actor.example"],
        description="""The uri being looked up""",
    )

    actor: str = Field(
        examples=["http://abel.example/actor"],
        description="""The actor performing the lookup""",
    )

    result: dict | None = Field(
        default=None,
        examples=[{"id": "http://actor.example", "type": "Person", "name": "Jane Doe"}],
        description="""The result of the lookup, None if no result yet,
    the result will be returned once the lookup is finished""",
    )

messages

CreateActorMessage

Bases: BaseModel

Parameters:

Name	Type	Description	Default
baseUrl	str	base url used to create the user on. Can contain a path	required
preferredUsername	str | None	Add a preferred username. This name will be used in acct:username@domain and supplied to webfinger. Here domain is determine from baseUrl.	None
profile	Dict[str, Any]	New profile object for the actor. The fields.	{}
autoFollow	bool	Enables setting actors to automatically accept follow requests	False

Source code in cattle_grid/model/messages.py

class CreateActorMessage(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
    )
    baseUrl: str = Field(
        examples=["http://local.example/"],
        description="""
    base url used to create the user on. Can contain a path
    """,
    )
    preferredUsername: str | None = Field(
        None,
        examples=["alice", "bob"],
        description="""
    Add a preferred username. This name will be used in acct:username@domain and supplied to webfinger. Here domain is determine from baseUrl.
    """,
    )
    profile: Dict[str, Any] = Field(
        {},
        examples=[{"summary": "A new actor"}],
        description="""
    New profile object for the actor. The fields.
    """,
    )
    autoFollow: bool = Field(
        False,
        examples=[True],
        description="""
    Enables setting actors to automatically accept follow requests
    """,
    )

processing

StoreActivityMessage

Bases: BaseModel

Stores the activity and then sends it, an id is assigned

Parameters:

Name	Type	Description	Default
actor	str		required
data	dict		required

Source code in cattle_grid/model/processing.py

class StoreActivityMessage(BaseModel):
    """Stores the activity and then sends it, an id is assigned"""

    actor: str
    data: dict

ToSendMessage

Bases: BaseModel

Internally used to send a message from actor to target with the content data

Parameters:

Name	Type	Description	Default
actor	str		required
data	dict		required
target	str		required

Source code in cattle_grid/model/processing.py

class ToSendMessage(BaseModel):
    """Internally used to send a message from actor
    to target with the content data"""

    actor: str
    data: dict
    target: str