.activity_pub.processing

cattle_grid.activity_pub.processing

incoming

incoming_accept_activity async

incoming_accept_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an incoming Accept activity

Source code in cattle_grid/activity_pub/processing/incoming.py

async def incoming_accept_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an incoming Accept activity"""
    accept_request = message.data
    request_being_accepted = id_for_object(accept_request.get("object"))

    following = await session.scalar(
        select(Following).where(Following.request == request_being_accepted)
    )

    if not following:
        logger.warning("Follow request with id '%s' not found", request_being_accepted)
        return

    following.accepted = True
    logger.info("Processed follow request %s (following)", request_being_accepted)

incoming_block_activity async

incoming_block_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an incoming Block activity

Source code in cattle_grid/activity_pub/processing/incoming.py

async def incoming_block_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an incoming Block activity"""
    current_actor_id = message.data.get("object")
    if current_actor_id != actor.actor_id:
        logger.warning("Mismatch of actor and target of block")
        return

    actor_blocking = message.data.get("actor")

    await session.execute(
        delete(Following).where(Following.following == actor_blocking)
    )

incoming_follow_request async

incoming_follow_request(
    message: ActivityMessage,
    actor: MessageActor,
    factories: FactoriesForActor,
    session: CommittingSession,
    publisher: InternalExchangePublisher,
)

For an incoming Follow request an entry in the Follower table is created with having accepted set to False.

If the actor automatically accepts followers, the actor sends Accept activity to the actor requesting to follow it.

Source code in cattle_grid/activity_pub/processing/incoming.py

async def incoming_follow_request(
    message: ActivityMessage,
    actor: MessageActor,
    factories: FactoriesForActor,
    session: CommittingSession,
    publisher: InternalExchangePublisher,
):
    """
    For an incoming Follow request an entry in the Follower table is created
    with having `accepted` set to False.

    If the actor automatically accepts followers, the actor sends Accept activity
    to the actor requesting to follow it.
    """
    follow_request = message.data
    to_follow = follow_request.get("object")
    if isinstance(to_follow, dict):
        to_follow = to_follow.get("id")

    if to_follow is None or to_follow != actor.actor_id:
        return

    request_id = follow_request.get("id")
    follower = follow_request.get("actor")

    await session.merge(
        Follower(
            actor=actor,
            follower=follower,
            request=request_id,
            accepted=False,
        )
    )

    if actor.automatically_accept_followers:
        accept = factories[0].accept(follow_request).build()

        await publisher(
            StoreActivityMessage(actor=actor.actor_id, data=accept),
            routing_key="store_activity",
        )
        logger.info(
            "Got follow request from %s with id %s (auto accepted)",
            follower,
            request_id,
        )

    else:
        logger.info("Got follow request from %s with id %s", follower, request_id)

incoming_reject_activity async

incoming_reject_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an incoming Reject activity

Source code in cattle_grid/activity_pub/processing/incoming.py

async def incoming_reject_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an incoming Reject activity"""
    reject_request = message.data
    request_being_rejected = reject_request.get("object")

    await session.execute(
        delete(Following).where(Following.request == request_being_rejected)
    )

incoming_undo_activity async

incoming_undo_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an incoming Undo activity

Source code in cattle_grid/activity_pub/processing/incoming.py

async def incoming_undo_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an incoming Undo activity"""

    accept_request = message.data
    request_being_undone = id_for_object(accept_request.get("object"))

    await session.execute(
        delete(Follower).where(Follower.request == request_being_undone)
    )

outgoing

outgoing_accept_request async

outgoing_accept_request(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an outgoing Accept activity

Source code in cattle_grid/activity_pub/processing/outgoing.py

async def outgoing_accept_request(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an outgoing Accept activity"""
    accept_request = message.data
    request_being_accepted = id_for_object(accept_request.get("object"))

    follower = await session.scalar(
        select(Follower).where(Follower.request == request_being_accepted)
    )

    if not follower:
        logger.warning("Follow request with id '%s' not found", request_being_accepted)
        return

    follower.accepted = True
    logger.info("Accepted follow request %s", request_being_accepted)

outgoing_block_activity async

outgoing_block_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an outgoing Block activity

Source code in cattle_grid/activity_pub/processing/outgoing.py

async def outgoing_block_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an outgoing Block activity"""
    block_request = message.data
    actor_being_blocked = block_request.get("object")

    await session.execute(
        delete(Follower).where(Follower.follower == actor_being_blocked)
    )

    block_id = block_request.get("id", "permanent")

    if block_id == "permanent":
        logger.warning("%s permanently blocked %s", actor.actor_id, actor_being_blocked)

    session.add(
        Blocking(
            actor=actor, blocking=actor_being_blocked, request=block_id, active=True
        )
    )
    logger.info("%s blocked %s", actor.actor_id, actor_being_blocked)

outgoing_follow_request async

outgoing_follow_request(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an outgoing Follow request

Source code in cattle_grid/activity_pub/processing/outgoing.py

async def outgoing_follow_request(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an outgoing Follow request"""

    follow_request = message.data
    to_follow = follow_request.get("object")
    if isinstance(to_follow, dict):
        to_follow = to_follow.get("id")

    if to_follow is None:
        return

    logger.info("Send follow request to %s", to_follow)

    session.add(
        Following(
            actor=actor,
            following=to_follow,
            request=follow_request.get("id"),
            accepted=False,
        )
    )

outgoing_message_distribution async

outgoing_message_distribution(
    message: ActivityMessage,
    session: SqlSession,
    publisher: InternalExchangePublisher,
)

Distributes the message to its recipients

Source code in cattle_grid/activity_pub/processing/outgoing.py

async def outgoing_message_distribution(
    message: ActivityMessage, session: SqlSession, publisher: InternalExchangePublisher
):
    """Distributes the message to its recipients"""

    recipients = recipients_for_object(message.data)
    recipients = remove_public(recipients)

    logger.debug("Got recipients %s", ", ".join(recipients))

    recipients = await update_recipients_for_collections(session, message, recipients)

    for recipient in recipients:
        await publisher(
            ToSendMessage(actor=message.actor, data=message.data, target=recipient),
            routing_key="to_send",
        )

outgoing_reject_activity async

outgoing_reject_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an outgoing Reject activity

Source code in cattle_grid/activity_pub/processing/outgoing.py

async def outgoing_reject_activity(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an outgoing Reject activity"""
    reject_request = message.data
    request_being_rejected = reject_request.get("object")

    await session.execute(
        delete(Follower).where(Follower.request == request_being_rejected)
    )

outgoing_undo_request async

outgoing_undo_request(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
)

Handles an outgoing Undo activity

Source code in cattle_grid/activity_pub/processing/outgoing.py

async def outgoing_undo_request(
    message: ActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
):
    """Handles an outgoing Undo activity"""
    accept_request = message.data
    request_being_undone = accept_request.get("object")

    await session.execute(
        delete(Following).where(Following.request == request_being_undone)
    )
    await session.execute(
        update(Blocking)
        .where(Blocking.request == request_being_undone)
        .values(active=False)
    )

remote

fetch_object async

fetch_object(
    message: FetchMessage,
    actor: MessageBovineActor,
    lookup: LookupAnnotation,
)

Handles retrieving a remote object

Source code in cattle_grid/activity_pub/processing/remote.py

async def fetch_object(
    message: FetchMessage,
    actor: MessageBovineActor,
    lookup: LookupAnnotation,
):
    """Handles retrieving a remote object"""

    try:
        lookup_result = await lookup(Lookup(uri=message.uri, actor=message.actor))
        if lookup_result.result:
            return lookup_result.result
    except Exception as e:
        logger.error("Something went up with lookup")
        logger.exception(e)

        lookup_result = Lookup(uri=message.uri, actor=message.actor)

    result = await actor.get(lookup_result.uri, fail_silently=True)

    return result

resolve_inbox async

resolve_inbox(
    session: AsyncSession, actor: BovineActor, target: str
)

Resolves the inbox of target for actor using a cache

Source code in cattle_grid/activity_pub/processing/remote.py

async def resolve_inbox(session: AsyncSession, actor: BovineActor, target: str):
    """Resolves the inbox of target for actor using
    a cache"""
    cached = await session.scalar(
        select(InboxLocation).where(InboxLocation.actor == target)
    )

    if cached:
        return cached.inbox

    target_actor = await actor.get(target)
    if not target_actor:
        return None

    inbox = target_actor.get("inbox")
    if inbox is None:
        return

    session.add(InboxLocation(actor=target, inbox=inbox))

    try:
        await session.commit()
    except IntegrityError:
        ...
    return inbox

sending_message async

sending_message(
    message: ToSendMessage,
    actor: MessageBovineActor,
    sql_session: SqlSession,
)

Handles sending a message

Source code in cattle_grid/activity_pub/processing/remote.py

async def sending_message(
    message: ToSendMessage, actor: MessageBovineActor, sql_session: SqlSession
):
    """Handles sending a message"""
    inbox = await resolve_inbox(sql_session, actor, message.target)
    if inbox:
        result = await actor.post(inbox, message.data)
        logger.info("Got %s for sending to %s", str(result), inbox)

shared_inbox

handle_shared_inbox_message async

handle_shared_inbox_message(
    message: SharedInboxMessage,
    session: SqlSession,
    broker: RabbitBroker = Context(),
)

This method is used to handle incoming messages from the shared inbox.

Source code in cattle_grid/activity_pub/processing/shared_inbox.py

async def handle_shared_inbox_message(
    message: SharedInboxMessage,
    session: SqlSession,
    broker: RabbitBroker = Context(),
):
    """
    This method is used to handle incoming messages from the shared inbox.
    """

    recipients = recipients_for_object(message.data)
    sender = message.data.get("actor")

    if sender is None:
        return

    local_actor_ids = to_result_set(
        await session.scalars(
            select(Actor.actor_id).where(Actor.actor_id.in_(recipients))
        )
    )
    following_actor_ids = {
        x.actor.actor_id
        for x in await session.scalars(
            select(Following)
            .where(Following.following == sender)
            .where(Following.accepted)
        )
    }

    for actor in local_actor_ids | following_actor_ids:
        await enqueue_from_inbox(
            broker, global_container.internal_exchange, actor, message.data
        )

store_activity

store_activity_subscriber async

store_activity_subscriber(
    message: StoreActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
    publisher: InternalExchangePublisher,
)

This method used internally to store activities generated by cattle_grid, e.g. an automatically generated accept requests. We note that it assigns an id to the activity.

After storing the activity, it is published to the appropriate outgoing.* topic, which triggers sending the activity.

Source code in cattle_grid/activity_pub/processing/store_activity.py

async def store_activity_subscriber(
    message: StoreActivityMessage,
    actor: MessageActor,
    session: CommittingSession,
    publisher: InternalExchangePublisher,
):
    """This method used internally to store activities
    generated by `cattle_grid`, e.g. an automatically generated
    accept requests. We note that it assigns an id to
    the activity.

    After storing the activity, it is published to the
    appropriate `outgoing.*` topic, which triggers sending
    the activity.
    """
    base_url = "/".join(actor.actor_id.split("/")[:-2])
    uuid = str(uuid7())
    activity_id = f"{base_url}/object/{uuid}"

    activity = message.data
    activity["id"] = activity_id
    activity_type = activity.get("type")

    session.add(
        StoredActivity(id=uuid, actor=actor, data=activity, published=datetime.now())
    )

    await publisher(
        {"actor": actor.actor_id, "data": activity},
        routing_key=f"outgoing.{activity_type}",
    )

testing

account_for_test async

account_for_test(sql_session) -> Account

Fixture to create an account

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture
async def account_for_test(sql_session) -> Account:
    """Fixture to create an account"""
    result = await create_account(sql_session, "alice", "alice", permissions=["admin"])
    assert result
    return result

actor_for_test async

actor_for_test(sql_session) -> Actor

Fixture to create an actor

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture
async def actor_for_test(sql_session) -> Actor:
    """Fixture to create an actor"""
    actor = await create_actor(sql_session, "http://localhost/ap")

    return actor

actor_with_account async

actor_with_account(sql_session, account_for_test) -> Actor

Fixture to create an actor with an account

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture
async def actor_with_account(sql_session, account_for_test) -> Actor:
    """Fixture to create an actor with an account"""
    actor = await create_actor(sql_session, "http://localhost/ap")
    await add_actor_to_account(
        sql_session, account_for_test, actor, name="test_fixture"
    )

    await sql_session.refresh(actor)

    return actor

loaded_config

loaded_config()

Ensures the configuration variables are loaded

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture(autouse=True, scope="session")
def loaded_config():
    """Ensures the configuration variables are loaded"""
    global_container.load_config()

sql_engine_for_tests async

sql_engine_for_tests()

Provides the sql engine (as in memory sqlite) for tests

This fixture has autouse=True, meaning that by importing

from cattle_grid.testing.fixtures import sql_engine_for_tests

it will run automatically. The engine is initialized in the place cattle_grid expects it.

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture(autouse=True)
async def sql_engine_for_tests():
    """Provides the sql engine (as in memory sqlite) for tests

    This fixture has autouse=True, meaning that by importing

    ```python
    from cattle_grid.testing.fixtures import sql_engine_for_tests
    ```

    it will run automatically. The engine is initialized in the
    place cattle_grid expects it.
    """
    async with global_container.alchemy_database(
        "sqlite+aiosqlite:///:memory:", echo=False
    ) as engine:
        async with engine.begin() as conn:
            await conn.run_sync(APBase.metadata.create_all)

        yield engine

sql_session async

sql_session(session_maker_for_tests)

Returns an AsyncSession to be used by tests

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture()
async def sql_session(session_maker_for_tests):
    """Returns an [AsyncSession][sqlalchemy.ext.asyncio.AsyncSession] to be used by tests"""
    async with session_maker_for_tests() as session:
        yield session

util

update_recipients_for_collections async

update_recipients_for_collections(
    session: AsyncSession,
    msg: ActivityMessage,
    recipients: Set[str],
) -> Set[str]

Updates recipients with followers and following collection.

Source code in cattle_grid/activity_pub/processing/util.py

async def update_recipients_for_collections(
    session: AsyncSession, msg: ActivityMessage, recipients: Set[str]
) -> Set[str]:
    """Updates recipients with followers and following collection."""

    db_actor = await session.scalar(select(Actor).where(Actor.actor_id == msg.actor))

    if db_actor is None:
        raise ValueError("Actor not found")

    self_delete = actor_deletes_themselves(msg.data)

    return await update_recipients_for_actor(
        session, db_actor, recipients, self_delete=self_delete
    )