cattle_hugs

object_router module-attribute

object_router = APIRouter()

Adds the API methods for the likes, replies, and shares collection. For this to work nginx needs to set the x-ap-location header, e.g.

auth_request /auth;
auth_request_set $requester $upstream_http_x_cattle_grid_requester;

proxy_set_header X-AP-Location "$scheme://$host$request_uri";
proxy_set_header X-cattle-grid-Requester $requester;

actor_for_id async

actor_for_id(
    session: AsyncSession,
    fetcher: Callable[[str], Awaitable[Actor | None]]
    | None,
    actor_id: str,
) -> Actor

Returns the Actor as a muck_out model

Source code in cattle_hugs/public/__init__.py

async def actor_for_id(
    session: AsyncSession,
    fetcher: Callable[[str], Awaitable[Actor | None]] | None,
    actor_id: str,
) -> Actor:
    """Returns the Actor as a muck_out model"""
    in_db = await db_actor_for_id(session, fetcher, actor_id)

    return Actor.model_validate(in_db.data)

db_actor_for_id async

db_actor_for_id(
    session: AsyncSession,
    fetcher: Callable[[str], Awaitable[Actor | None]]
    | None,
    actor_id: str,
) -> StoredActor

Returns the database object for the actor

Source code in cattle_hugs/public/__init__.py

async def db_actor_for_id(
    session: AsyncSession,
    fetcher: Callable[[str], Awaitable[Actor | None]] | None,
    actor_id: str,
) -> StoredActor:
    """Returns the database object for the actor"""
    in_db = await session.scalar(select(StoredActor).where(StoredActor.id == actor_id))

    if in_db:
        return in_db

    if fetcher is None:
        raise ActorNotFoundError("No fetcher provided")

    result = await fetcher(actor_id)

    if result is None:
        raise ActorNotFoundError(f"Actor not found: {actor_id}")

    actor_to_store = StoredActor(
        id=result.id, data=result.model_dump(exclude_none=True)
    )
    session.add(actor_to_store)
    await session.commit()
    await session.refresh(actor_to_store)

    return actor_to_store

register_base_object async

register_base_object(session: AsyncSession, obj: dict)

Adds an obj to be tracked

Source code in cattle_hugs/public/objects.py

async def register_base_object(session: AsyncSession, obj: dict):
    """Adds an obj to be tracked"""
    object_id = get_object_id(obj)

    in_db = await session.scalar(
        select(BaseObject).where(BaseObject.object_id == object_id)
    )
    if in_db:
        return in_db

    base_object = BaseObject(object_id=object_id)
    session.add(base_object)
    await session.commit()

    return base_object

retrieve_interactions async

retrieve_interactions(
    session: AsyncSession, object_id: str
) -> None | Interactions

Returns the interactions

Source code in cattle_hugs/public/interactions.py

async def retrieve_interactions(
    session: AsyncSession, object_id: str
) -> None | Interactions:
    """Returns the interactions"""
    result = await session.scalar(
        select(BaseObject)
        .where(BaseObject.object_id == object_id)
        .options(joinedload(BaseObject.comments))
        .options(joinedload(BaseObject.interactions))
    )

    if not result:
        return

    comments = [
        CommentInfo(data=x.data)
        for x in result.comments
        if x.status != CommentStatus.deleted
    ]
    interactions = [
        InteractionInfo(interaction_type=x.interaction_type, data=x.activity)
        for x in result.interactions
    ]

    actors_for_id = {}
    for x in result.comments:
        actors_for_id[x.actor.id] = Actor.model_validate(x.actor.data)
    for x in result.interactions:
        actors_for_id[x.actor.id] = Actor.model_validate(x.actor.data)

    actors = list(actors_for_id.values())

    return Interactions(
        object_id=object_id, comments=comments, interactions=interactions, actors=actors
    )

with_collections

with_collections(
    obj: dict,
    collections: list[str] = ["replies", "shares", "likes"],
)

Adds the collections to the object

Source code in cattle_hugs/public/objects.py

def with_collections(
    obj: dict, collections: list[str] = ["replies", "shares", "likes"]
):
    """Adds the collections to the object"""
    object_id = get_object_id(obj)

    result = {**obj}
    for collection in collections:
        result[collection] = f"{object_id}/{collection}"

    return result