Skip to content

cattle_grid.extensions.examples.html_display

The goal of this extension is to illustrate the mechanism how cattle_grid can be used to create a webpage displaying the public posts of its users. The focus is on providing a working solution to key problems than on providing the optimal solution.

An example of a feed is available at @release. The current scope is:

  • Display actor and posts
  • Correct redirecting behavior for easy use
  • Display replies (todo)
  • Export
  • Import from compatible sources (todo)

This is not an independent solution. It relies on both the context and simple_storage extension of cattle_grid.

export_create_token async

export_create_token(
    publishing_actor: PublishingActor,
    session: CommittingSession,
    publisher: AccountExchangePublisher,
    config: Config,
)

Provides a token and download link for the export of stored objects

The response is of type cattle_grid.extensions.examples.html_display.types.ExportTokenResponse and send to receive.NAME.response.trigger.

Source code in cattle_grid/extensions/examples/html_display/__init__.py
@extension.subscribe("html_display_export", replies=True)
async def export_create_token(
    publishing_actor: PublishingActor,
    session: CommittingSession,
    publisher: AccountExchangePublisher,
    config: extension.Config,  # type:ignore
):
    """Provides a token and download link for the export of stored objects

    The response is of type [cattle_grid.extensions.examples.html_display.types.ExportTokenResponse][] and
    send to `receive.NAME.response.trigger`."""
    token_uuid = uuid4()
    session.add(
        ExportPermission(publishing_actor=publishing_actor, one_time_token=token_uuid)
    )

    account = await account_for_actor(session, publishing_actor.actor)

    if not account:
        raise Exception("Could not find account")

    await publisher(
        ExportTokenResponse(
            actor=publishing_actor.actor,
            token=str(token_uuid),
            export_url=config.html_url(publishing_actor.actor, publishing_actor.name)
            + f"/export?token={str(token_uuid)}",
        ),
        routing_key=f"receive.{account.name}.response.trigger",
    )

html_display_incoming_announce async

html_display_incoming_announce(
    activity: ParsedActivity,
    message: ActivityMessage,
    session: CommittingSession,
)

Handles announces

Source code in cattle_grid/extensions/examples/html_display/__init__.py
@extension.subscribe("incoming.Announce")
async def html_display_incoming_announce(
    activity: ParsedActivity,
    message: ActivityMessage,
    session: CommittingSession,
):
    """Handles announces"""
    if activity is None:
        logger.info("Unparsed announce: " + json.dumps(message.data, indent=2))
        return
    object_id = activity.object

    if not isinstance(object_id, str):
        return

    published_object = await object_for_object_id(session, object_id)

    session.add(
        PublishedObjectInteraction(
            id=uuid7(),
            published_object=published_object,
            object_id=activity.id,
            interaction=InteractionType.shares,
        )
    )

html_display_incoming_create async

html_display_incoming_create(
    embedded_object: ParsedEmbeddedObject,
    session: CommittingSession,
)

Handles replies

Source code in cattle_grid/extensions/examples/html_display/__init__.py
@extension.subscribe("incoming.Create")
async def html_display_incoming_create(
    embedded_object: ParsedEmbeddedObject,
    session: CommittingSession,
):
    """Handles replies"""
    if not embedded_object:
        return

    object_id = embedded_object.in_reply_to

    if not isinstance(object_id, str):
        return

    published_object = await object_for_object_id(session, object_id)

    session.add(
        PublishedObjectInteraction(
            id=uuid7(),
            published_object=published_object,
            object_id=embedded_object.id,
            interaction=InteractionType.replies,
        )
    )

html_display_incoming_like async

html_display_incoming_like(
    activity: ParsedActivity,
    message: ActivityMessage,
    session: CommittingSession,
)

Handles likes

Source code in cattle_grid/extensions/examples/html_display/__init__.py
@extension.subscribe("incoming.Like")
async def html_display_incoming_like(
    activity: ParsedActivity,
    message: ActivityMessage,
    session: CommittingSession,
):
    """Handles likes"""
    if activity is None:
        logger.info("Unparsed like: " + json.dumps(message.data, indent=2))
        return
    object_id = activity.object

    if not isinstance(object_id, str):
        return

    published_object = await object_for_object_id(session, object_id)

    session.add(
        PublishedObjectInteraction(
            id=uuid7(),
            published_object=published_object,
            object_id=activity.id,
            interaction=InteractionType.likes,
        )
    )

html_publish_object async

html_publish_object(
    message: ActivityMessage,
    session: CommittingSession,
    publishing_actor: PublishingActor,
    config: Config,
    factories: FactoriesForActor,
    activity_publisher: ActivityExchangePublisher,
)

Publishes an object

Source code in cattle_grid/extensions/examples/html_display/__init__.py
@extension.subscribe("html_display_publish_object")
async def html_publish_object(
    message: ActivityMessage,
    session: CommittingSession,
    publishing_actor: PublishingActor,
    config: extension.Config,  # type:ignore
    factories: FactoriesForActor,
    activity_publisher: ActivityExchangePublisher,
):
    """Publishes an object"""
    obj = message.data

    if not is_public(obj):
        await activity_publisher(
            ActivityMessage(actor=message.actor, data=obj),
            routing_key="publish_object",
        )
        return

    if obj.get("id"):
        raise ValueError("Object ID must not be set")

    if obj.get("attributedTo") != message.actor:
        raise ValueError("Actor must match object attributedTo")

    publisher = Publisher(publishing_actor, config, obj)
    session.add(
        PublishedObject(
            id=publisher.uuid,
            data=publisher.object_for_store,
            actor=publishing_actor.actor,
        )
    )

    activity = factories[0].create(publisher.object_for_remote).build()

    await activity_publisher(
        ActivityMessage(actor=message.actor, data=activity),
        routing_key="publish_activity",
    )

name_actor async

name_actor(
    message: NameActorMessage,
    publishing_actor: PublishingActor,
    session: CommittingSession,
    activity_publisher: ActivityExchangePublisher,
    config: Config,
)

Sets the display name of the actor

Source code in cattle_grid/extensions/examples/html_display/__init__.py
@extension.subscribe("html_display_name")
async def name_actor(
    message: NameActorMessage,
    publishing_actor: PublishingActor,
    session: CommittingSession,
    activity_publisher: ActivityExchangePublisher,
    config: extension.Config,  # type:ignore
):
    """Sets the display name of the actor"""
    if message.actor != publishing_actor.actor:
        raise Exception("Actor mismatch")

    publishing_actor.name = message.name

    await activity_publisher(
        UpdateActorMessage(
            actor=publishing_actor.actor,
            actions=[
                UpdateUrlAction(
                    action=UpdateActionType.add_url,
                    url=config.html_url(publishing_actor.actor, publishing_actor.name),
                    media_type="text/html",
                )
            ],
        ),
        routing_key="update_actor",
    )

    if config.automatically_add_users_to_group:
        manager = ActorManager(actor_id=publishing_actor.actor, session=session)
        await manager.add_to_group("html_display")

outgoing_delete async

outgoing_delete(
    message: ActivityMessage, session: SqlSession
)

Deletes the publishing actor, if they delete themself

Source code in cattle_grid/extensions/examples/html_display/__init__.py
@extension.subscribe("outgoing.Delete")
async def outgoing_delete(
    message: ActivityMessage,
    session: SqlSession,
):
    """Deletes the publishing actor, if they delete themself"""
    activity = message.data.get("raw")
    if not isinstance(activity, dict):
        return
    if not actor_deletes_themselves(activity):
        return

    actor = await session.scalar(
        select(DBPublishingActor).where(DBPublishingActor.actor == message.actor)
    )
    if not actor:
        return

    logger.info("Deleting publishing actor with name %s", actor.name)

    await session.delete(actor)
    await session.flush()

config

HtmlDisplayConfiguration

Bases: BaseModel

Parameters:

Name Type Description Default
prefix str

Path to use before the generated uuid. The protocol and domain will be extracted from the actor id. See determine_url_start.

'/html_display/object/'
html_prefix str

Prefix to use for the url pages

'/@'
automatically_add_users_to_group bool

Causes an actor that sends a html_display_name message to be added to the html_display group

False
Source code in cattle_grid/extensions/examples/html_display/config.py
class HtmlDisplayConfiguration(BaseModel):
    prefix: str = Field(
        default="/html_display/object/",
        description="Path to use before the generated uuid. The protocol and domain will be extracted from the actor id. See [determine_url_start][cattle_grid.extensions.examples.simple_storage.config.determine_url_start].",
    )

    html_prefix: str = Field(
        default="/@", description="Prefix to use for the url pages"
    )

    automatically_add_users_to_group: bool = Field(
        default=False,
        description="Causes an actor that sends a html_display_name message to be added to the html_display group",
    )

    def url_start(self, actor_id):
        return determine_url_start(actor_id, self.prefix)

    def html_url_start(self, actor_id):
        """
        Determines the start of an url

        ```python
        >>> config = HtmlDisplayConfiguration()
        >>> config.html_url_start("http://actor.example/some/id")
        'http://actor.example/@'

        ```
        """
        return determine_url_start(actor_id, self.html_prefix)

    def html_url(self, actor_id, actor_name):
        """
        Determines the start of an url

        ```python
        >>> config = HtmlDisplayConfiguration()
        >>> config.html_url("http://actor.example/some/id", "john")
        'http://actor.example/@john'

        ```
        """
        return self.html_url_start(actor_id) + actor_name

html_url

html_url(actor_id, actor_name)

Determines the start of an url

>>> config = HtmlDisplayConfiguration()
>>> config.html_url("http://actor.example/some/id", "john")
'http://actor.example/@john'
Source code in cattle_grid/extensions/examples/html_display/config.py
def html_url(self, actor_id, actor_name):
    """
    Determines the start of an url

    ```python
    >>> config = HtmlDisplayConfiguration()
    >>> config.html_url("http://actor.example/some/id", "john")
    'http://actor.example/@john'

    ```
    """
    return self.html_url_start(actor_id) + actor_name

html_url_start

html_url_start(actor_id)

Determines the start of an url

>>> config = HtmlDisplayConfiguration()
>>> config.html_url_start("http://actor.example/some/id")
'http://actor.example/@'
Source code in cattle_grid/extensions/examples/html_display/config.py
def html_url_start(self, actor_id):
    """
    Determines the start of an url

    ```python
    >>> config = HtmlDisplayConfiguration()
    >>> config.html_url_start("http://actor.example/some/id")
    'http://actor.example/@'

    ```
    """
    return determine_url_start(actor_id, self.html_prefix)

database

Base

Bases: AsyncAttrs, DeclarativeBase

Base model

Source code in cattle_grid/extensions/examples/html_display/database.py
class Base(AsyncAttrs, DeclarativeBase):
    """Base model"""

    pass

ExportPermission

Bases: Base

Records a one time token required to download the export

Source code in cattle_grid/extensions/examples/html_display/database.py
class ExportPermission(Base):
    """Records a one time token required to download the export"""

    __tablename__ = "html_display_export_permission"

    id: Mapped[int] = mapped_column(primary_key=True)
    publishing_actor_id: Mapped[str] = mapped_column(
        ForeignKey("html_display_publishing_actor.id")
    )
    publishing_actor: Mapped[PublishingActor] = relationship()
    one_time_token: Mapped[bytes] = mapped_column(UUIDType(binary=True))

PublishedObject

Bases: Base

HTML display object in the database

Source code in cattle_grid/extensions/examples/html_display/database.py
class PublishedObject(Base):
    """HTML display object in the database"""

    __tablename__ = "html_display_stored_object"

    id: Mapped[bytes] = mapped_column(UUIDType(binary=True), primary_key=True)
    data: Mapped[dict] = mapped_column(JSON)
    actor: Mapped[str] = mapped_column()
    create_date: Mapped[datetime.datetime] = mapped_column(server_default=func.now())

    interactions: Mapped[list["PublishedObjectInteraction"]] = relationship(
        viewonly=True
    )

PublishedObjectInteraction

Bases: Base

HTML display object in the database

Source code in cattle_grid/extensions/examples/html_display/database.py
class PublishedObjectInteraction(Base):
    """HTML display object in the database"""

    __tablename__ = "html_display_stored_object_interaction"

    id: Mapped[bytes] = mapped_column(UUIDType(binary=True), primary_key=True)

    published_object_id: Mapped[str] = mapped_column(
        ForeignKey("html_display_stored_object.id")
    )
    published_object: Mapped[PublishedObject] = relationship()

    object_id: Mapped[str] = mapped_column()

    interaction: Mapped[InteractionType] = mapped_column(Enum(InteractionType))

dependencies

PublishingActor module-attribute

PublishingActor = Annotated[
    PublishingActor, Depends(publishing_actor)
]

Returns the publishing actor

fastapi_dependencies

ActivityPubActor module-attribute

ActivityPubActor = Annotated[Actor, Depends(get_actor)]

Returns the Actor from the database

ActorProfile module-attribute

ActorProfile = Annotated[dict, Depends(get_actor_profile)]

Returns the actor profile

PublishedObjectForUUID module-attribute

PublishedObjectForUUID = Annotated[
    PublishedObject, Depends(published_object)
]

Returns the published object

PublishedObjectWithInteractionsForUUID module-attribute

PublishedObjectWithInteractionsForUUID = Annotated[
    PublishedObject,
    Depends(published_object_With_interactions),
]

Returns the published object, joined with the interactions, i.e.

async def func(obj: PublishedObjectWithInteractionsForUUID):
    for x in obj.interactions:
        print(x)

is valid pseudo code.

PublishingActorForName module-attribute

PublishingActorForName = Annotated[
    PublishingActor, Depends(publishing_actor_for_name)
]

Returns the publishing actor

publisher

Publisher dataclass

Class for manipulating objects being published

Parameters:

Name Type Description Default
actor PublishingActor
required
config HtmlDisplayConfiguration
required
obj dict[str, Any]
required
uuid UUID

UUID version 7 features a time-ordered value field derived from the widely implemented and well known Unix Epoch timestamp source, the number of milliseconds since midnight 1 Jan 1970 UTC, leap seconds excluded. As well as improved entropy characteristics over versions 1 or 6.

Implementations SHOULD utilize UUID version 7 over UUID version 1 and 6 if possible.

UUID('019a4555-5a71-7c7c-9fc9-41d84efcec6f')
Source code in cattle_grid/extensions/examples/html_display/publisher.py
@dataclass
class Publisher:
    """Class for manipulating objects being published"""

    actor: PublishingActor
    config: HtmlDisplayConfiguration
    obj: dict[str, Any]
    uuid: UUID = field(default_factory=uuid7)

    def __post_init__(self):
        if "id" in self.obj:
            obj_id = self.obj["id"]
            self.uuid = UUID(obj_id.split("/")[-1])
        else:
            self.obj = {
                **self.obj,
                "id": self.config.url_start(self.actor.actor) + str(self.uuid),
            }

    @property
    def object_for_store(self):
        return self.obj

    @property
    def object_for_remote(self):
        copy = {**self.obj}
        copy.update(self._collection_links())
        return self._add_url_to_obj(copy)

    def _collection_links(self):
        object_id = self.obj["id"]
        return {
            collection: f"{object_id}/{collection}"
            for collection in ["replies", "shares", "likes"]
        }

    def _add_url_to_obj(self, obj: dict):
        url_list = as_list(obj.get("url", []))

        url = (
            self.config.html_url_start(self.actor.actor)
            + self.actor.name
            + "/o/"
            + str(self.uuid)
        )

        obj["url"] = url_list + [self.create_html_link(url)]

        return obj

    def create_html_link(self, url: str | None = None):
        if url is None:
            url = self.actor.actor_id + "#html"
        return {"type": "Link", "mediaType": "text/html", "href": url}

router

export_objects async

export_objects(
    actor: PublishingActorForName,
    token: str,
    session: SqlSession,
)

Returns the export of user data, a token needs to be provided. The token can be requested via the html_display_export method.

Source code in cattle_grid/extensions/examples/html_display/router.py
@router.get("/html/{actor_name}/export", tags=["html"])
async def export_objects(
    actor: PublishingActorForName, token: str, session: SqlSession
):
    """Returns the export of user data, a token needs to be provided.
    The token can be requested via the `html_display_export` method."""
    try:
        token_uuid = UUID(token)
    except ValueError:
        raise HTTPException(422, detail="Invalid uuid as token")

    permission = await session.scalar(
        select(ExportPermission)
        .where(ExportPermission.publishing_actor == actor)
        .where(ExportPermission.one_time_token == token_uuid)
    )

    if not permission:
        raise HTTPException(401)

    items = await session.scalars(
        select(PublishedObject.data).where(PublishedObject.actor == actor.actor)
    )

    return OrderedCollection(id="objects.json", items=items.all()).build()  # type: ignore

get_actor_html async

get_actor_html(
    actor: PublishingActorForName,
    profile: ActorProfile,
    request: Request,
    session: SqlSession,
    should_serve: ShouldServe,
)

Returns the HTML representation of the actor

Source code in cattle_grid/extensions/examples/html_display/router.py
@router.get("/html/{actor_name}", response_class=HTMLResponse, tags=["html"])
@router.get("/html/{actor_name}/", response_class=HTMLResponse, tags=["html"])
async def get_actor_html(
    actor: PublishingActorForName,
    profile: ActorProfile,
    request: Request,
    session: SqlSession,
    should_serve: ShouldServe,
):
    """Returns the HTML representation of the actor"""

    if (
        ContentType.html not in should_serve
        and ContentType.activity_pub in should_serve
    ):
        return RedirectResponse(profile["id"])

    published_objects = await session.scalars(
        select(PublishedObject)
        .where(PublishedObject.actor == actor.actor)
        .order_by(PublishedObject.create_date.desc())
        .limit(10)
    )

    posts = [
        {
            "body": sanitize_html(x.data.get("content")),
            "date": x.data.get("published"),
            "id": str(x.id),
        }
        for x in published_objects
    ]

    return templates.TemplateResponse(
        request,
        name="index.html.j2",
        context={
            "name": actor.name,
            "profile": format_actor_profile(profile),
            "posts": posts,
            "version": __version__,
        },
    )

get_index async

get_index()

Dummy index page, only useful for debugging

Source code in cattle_grid/extensions/examples/html_display/router.py
@router.get("/")
async def get_index():
    """Dummy index page, only useful for debugging"""
    return "html extension"

get_object async

get_object(
    obj: PublishedObjectForUUID,
    ap_headers: ActivityPubHeaders,
    session: SqlSession,
    config: ConfigFastAPI,
)

Returns the stored object

Source code in cattle_grid/extensions/examples/html_display/router.py
@router.get("/object/{uuid}", response_class=ActivityResponse, tags=["activity_pub"])
async def get_object(
    obj: PublishedObjectForUUID,
    ap_headers: ActivityPubHeaders,
    session: SqlSession,
    config: extension.ConfigFastAPI,  # type:ignore
):
    """Returns the stored object"""
    if not ap_headers.x_cattle_grid_requester:
        raise HTTPException(401)

    if ap_headers.x_ap_location != obj.data.get("id"):
        raise HTTPException(404)

    if not await is_valid_requester_for_obj(
        session, ap_headers.x_cattle_grid_requester, obj.data
    ):
        raise HTTPException(401)

    actor = await session.scalar(
        select(PublishingActor).where(PublishingActor.actor == obj.actor)
    )
    if actor is None:
        raise HTTPException(404)

    publisher = Publisher(actor, config, obj=obj.data)

    return publisher.object_for_remote

get_object_html async

get_object_html(
    actor: PublishingActorForName,
    obj: PublishedObjectWithInteractionsForUUID,
    request: Request,
    should_serve: ShouldServe,
)

Returns the HTML representation for an object

Source code in cattle_grid/extensions/examples/html_display/router.py
@router.get("/html/{actor_name}/o/{uuid}", response_class=HTMLResponse, tags=["html"])
@router.get(
    "/html/{actor_name}/{uuid}",
    response_class=HTMLResponse,
    deprecated=True,
    tags=["html"],
)
async def get_object_html(
    actor: PublishingActorForName,
    obj: PublishedObjectWithInteractionsForUUID,
    request: Request,
    should_serve: ShouldServe,
):
    """Returns the HTML representation for an object"""
    if actor.actor != obj.actor:
        raise HTTPException(404)

    if (
        ContentType.html not in should_serve
        and ContentType.activity_pub in should_serve
    ):
        return RedirectResponse(obj.data["id"])

    interactions = {
        str(interaction): [
            x.object_id for x in obj.interactions if x.interaction == interaction
        ]
        for interaction in [
            InteractionType.shares,
            InteractionType.likes,
            InteractionType.replies,
        ]
    }

    return templates.TemplateResponse(
        request,
        name="object.html.j2",
        context={
            "name": actor.name,
            "content": sanitize_html(obj.data["content"]),
            **interactions,
            "version": __version__,
        },
    )

get_object_interaction async

get_object_interaction(
    obj: PublishedObjectWithInteractionsForUUID,
    ap_headers: ActivityPubHeaders,
    session: SqlSession,
    interaction: InteractionType,
)

Returns the stored object

Source code in cattle_grid/extensions/examples/html_display/router.py
@router.get(
    "/object/{uuid}/{interaction}",
    response_class=ActivityResponse,
    tags=["activity_pub"],
)
async def get_object_interaction(
    obj: PublishedObjectWithInteractionsForUUID,
    ap_headers: ActivityPubHeaders,
    session: SqlSession,
    interaction: InteractionType,
):
    """Returns the stored object"""
    if not ap_headers.x_cattle_grid_requester:
        raise HTTPException(401)

    collection_id = f"{obj.data.get('id')}/{interaction}"

    if ap_headers.x_ap_location != collection_id:
        raise HTTPException(404)

    if not await is_valid_requester_for_obj(
        session, ap_headers.x_cattle_grid_requester, obj.data
    ):
        raise HTTPException(401)

    logger.info(obj.interactions)

    return OrderedCollection(
        id=collection_id,
        items=[x.object_id for x in obj.interactions if x.interaction == interaction],
    ).build()

types

ExportTokenResponse

Bases: WithActor

Message containing the export information

Parameters:

Name Type Description Default
actor str

actor_id of the actor that received the message

required
token str

One time token

required
export_url str

Url including the token the export is located at

required
Source code in cattle_grid/extensions/examples/html_display/types.py
class ExportTokenResponse(WithActor):
    """Message containing the export information"""

    token: str = Field(description="One time token")
    export_url: str = Field(
        description="Url including the token the export is located at"
    )

NameActorMessage

Bases: WithActor

Message for renaming an actor

Parameters:

Name Type Description Default
actor str

actor_id of the actor that received the message

required
name str

Name for the actor

required
Source code in cattle_grid/extensions/examples/html_display/types.py
6
7
8
9
class NameActorMessage(WithActor):
    """Message for renaming an actor"""

    name: str = Field(description="Name for the actor", examples=["john"])