Skip to content

cattle_grid.model

cattle_grid.model

ActivityMessage

Bases: BaseModel

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
required
data Dict[str, Any]
required
Source code in cattle_grid/model/__init__.py
class ActivityMessage(BaseModel):
    """
    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",
    )
    actor: str
    """
    actor_id of the actor that received the message
    """
    data: Dict[str, Any]
    """
    Activity.
    """

actor instance-attribute

actor_id of the actor that received the message

data instance-attribute

Activity.

FetchMessage

Bases: BaseModel

Used to request an ActivityPub object to be retrieved

Parameters:

Name Type Description Default
actor str
required
uri str
required
Source code in cattle_grid/model/__init__.py
class FetchMessage(BaseModel):
    """
    Used to request an ActivityPub object to be retrieved
    """

    model_config = ConfigDict(
        extra="forbid",
    )
    actor: str
    """
    actor_id of the actor that received the message
    """
    uri: str
    """
    URI of the object being retrieved
    """

actor instance-attribute

actor_id of the actor that received the message

uri instance-attribute

URI of the object being retrieved

account

Objects used for the account exchange

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
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}`""",
    )

    preferred_username: 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.
    """,
        serialization_alias="preferredUsername",
    )
    profile: Dict[str, Any] = Field(
        {},
        examples=[{"summary": "A new actor"}],
        description="""
    New profile object for the actor.
    """,
    )
    automatically_accept_followers: bool | None = Field(
        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 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: 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

Send on outgoing or incoming events

Parameters:

Name Type Description Default
actor str

The actor performing the action

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
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/account.py
class EventInformation(WithActor):
    """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.""",
    )

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

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()

FetchMessage

Bases: WithActor

Message to fetch an object from the Fediverse

Parameters:

Name Type Description Default
actor str

The actor performing the action

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

The actor performing the action

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
actors List[str]

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[MethodInformation]

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

    actors: List[str] = Field(
        examples=[
            [
                "http://host.example/actor/1",
                "http://host.example/actor/2",
            ]
        ],
        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[MethodInformation] = Field(
        default_factory=list,
        examples=[
            [
                MethodInformation(
                    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

The actor performing the action

required
Source code in cattle_grid/model/account.py
class TriggerMessage(WithActor):
    """Message to trigger something on the ActivityExchange"""

    model_config = ConfigDict(extra="allow")

WithActor

Bases: BaseModel

Used as base for messages requiring an actor

Parameters:

Name Type Description Default
actor str

The actor performing the action

required
Source code in cattle_grid/model/account.py
class WithActor(BaseModel):
    """Used as base for messages requiring an actor"""

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

exchange

DeleteActorMessage

Bases: BaseModel

Allows one to delete the actor object

Parameters:

Name Type Description Default
actor str

The URI of the actor being deleted.

required
Source code in cattle_grid/model/exchange.py
class DeleteActorMessage(BaseModel):
    """
    Allows one to delete the actor object
    """

    model_config = ConfigDict(
        extra="forbid",
    )
    actor: str = Field(
        ...,
        examples=["http://local.example/actor"],
        description="""
    The URI of the actor being deleted.
    """,
    )

UpdateAction

Bases: BaseModel

Action to update an actor

Parameters:

Name Type Description Default
action UpdateActionType
required
Source code in cattle_grid/model/exchange.py
class UpdateAction(BaseModel):
    """Action to update an actor"""

    model_config = ConfigDict(
        extra="allow",
    )

    action: UpdateActionType

UpdateActorMessage

Bases: BaseModel

Allows one to update the actor object

Parameters:

Name Type Description Default
actor str

The URI of the actor being updated. Must be managed by cattle_grid

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[UpdateAction]

Actions to be taken when updating the profile

<dynamic>
Source code in cattle_grid/model/exchange.py
class UpdateActorMessage(BaseModel):
    """
    Allows one to update the actor object
    """

    # model_config = ConfigDict(
    #     extra="forbid",
    # )
    actor: str = Field(
        ...,
        examples=["http://local.example/actor"],
        description="""
    The URI of the actor being updated. Must be managed by cattle_grid
    """,
    )
    profile: Dict[str, Any] | None = Field(
        None,
        examples=[{"summary": "A new description of the actor"}],
        description="""
    New profile object for the actor. The fields.
    """,
    )
    autoFollow: bool | None = Field(
        None,
        examples=[True, False, None],
        description="""
    Enables setting actors to automatically accept follow requests
    """,
    )

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

UpdateIdentifierAction

Bases: UpdateAction

Used to update an identifier of the actor

Parameters:

Name Type Description Default
action UpdateActionType
required
identifier str
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.py
class UpdateIdentifierAction(UpdateAction):
    """Used to update an identifier of the actor"""

    identifier: str
    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",
    )

extension

MethodInformation

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
Source code in cattle_grid/model/extension.py
class MethodInformation(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(
        None,
        examples=["Send a message as the actor"],
        description="""Description of the method""",
    )

lookup

LookupMethod = Callable[[Lookup], Awaitable[Lookup]] module-attribute

Alias for the Lookup Method

Lookup

Bases: BaseModel

Lookup of something from the Fediverse

Parameters:

Name Type Description Default
uri str
required
actor str
required
result dict | None
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"])
    """The uri being looked up"""

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

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

The actor performing the lookup

result = Field(None, examples=[{'id': 'http://actor.example', 'type': 'Person', 'name': 'Jane Doe'}]) class-attribute instance-attribute

The result of the lookup, None if no result yet, the result will be returned once the lookup is finished

uri = Field(examples=['http://actor.example', 'acct:user@actor.example']) class-attribute instance-attribute

The uri being looked up

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