Skip to content

.activity_pub.server

cattle_grid.activity_pub.server

This package contains the overall router for all connection needs to the Fediverse. This means the .well-known endpoints.

webfinger_responder(resource) async

Handles requests to .well-known/webfinger. Results are determined by the identifier property of [PublicIdentifier][cattle_grid.ap.models.PublicIdentifier] matching the resource parameter.

See RFC 7033 WebFinger.

Source code in cattle_grid/activity_pub/server/__init__.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@router.get("/.well-known/webfinger")
async def webfinger_responder(resource: str) -> JrdData:
    """Handles requests to .well-known/webfinger. Results are determined by the identifier property of [PublicIdentifier][cattle_grid.ap.models.PublicIdentifier] matching the resource
    parameter.

    See [RFC 7033 WebFinger](https://www.rfc-editor.org/rfc/rfc7033).
    """

    logger.info("looking up web finger for resource '%s'", resource)

    pi = await PublicIdentifier.get_or_none(identifier=resource)
    if not pi:
        raise HTTPException(status_code=404, detail="Item not found")

    await pi.fetch_related("actor")

    return webfinger_response(pi.identifier, pi.actor.actor_id)

cattle_grid.activity_pub.server.router

ActivityPub related functionality

APHeaders

Bases: BaseModel

Headers every request should have. These should be added by the remote proxy.

Source code in cattle_grid/activity_pub/server/router.py 
23
24
25
26
27
28
29
class APHeaders(BaseModel):
    """Headers every request should have. These should be added by the remote proxy."""

    x_cattle_grid_requester: str
    """URI of the actor making the request"""
    x_ap_location: str
    """URI of the resource being retrieved"""

x_ap_location: str instance-attribute

URI of the resource being retrieved

x_cattle_grid_requester: str instance-attribute

URI of the actor making the request

ActivityResponse

Bases: JSONResponse

Response that ensures the content-type is “application/activity+json”

Source code in cattle_grid/activity_pub/server/router.py 
32
33
34
35
36
37
class ActivityResponse(JSONResponse):
    """Response that ensures the content-type is
    "application/activity+json"
    """

    media_type = "application/activity+json"

actor_profile(id_str, headers) async

Returns the actor

Source code in cattle_grid/activity_pub/server/router.py 
49
50
51
52
53
54
55
@ap_router.get("/actor/{id_str}", response_class=ActivityResponse)
async def actor_profile(id_str, headers: Annotated[APHeaders, Header()]):
    """Returns the actor"""
    actor = validate_actor(await Actor.get_or_none(actor_id=headers.x_ap_location))

    result = actor_to_object(actor)
    return result

followers(id_str, headers) async

Returns the followers

Source code in cattle_grid/activity_pub/server/router.py 
73
74
75
76
77
78
@ap_router.get("/followers/{id_str}", response_class=ActivityResponse)
async def followers(id_str, headers: Annotated[APHeaders, Header()]):
    """Returns the followers"""
    actor = validate_actor(await Actor.get_or_none(followers_uri=headers.x_ap_location))
    followers = await followers_for_actor(actor)
    return OrderedCollection(id=headers.x_ap_location, items=followers).build()

following(id_str, headers) async

Returns the following

Source code in cattle_grid/activity_pub/server/router.py 
65
66
67
68
69
70
@ap_router.get("/following/{id_str}", response_class=ActivityResponse)
async def following(id_str, headers: Annotated[APHeaders, Header()]):
    """Returns the following"""
    actor = validate_actor(await Actor.get_or_none(following_uri=headers.x_ap_location))
    following = await following_for_actor(actor)
    return OrderedCollection(id=headers.x_ap_location, items=following).build()

outbox(id_str, headers) async

Returns an empty ordered collection as outbox

Source code in cattle_grid/activity_pub/server/router.py 
58
59
60
61
62
@ap_router.get("/outbox/{id_str}", response_class=ActivityResponse)
async def outbox(id_str, headers: Annotated[APHeaders, Header()]):
    """Returns an empty ordered collection as outbox"""
    validate_actor(await Actor.get_or_none(outbox_uri=headers.x_ap_location))
    return OrderedCollection(id=headers.x_ap_location).build()