foxhole/app/actor.py

#!/usr/bin/env python3
import typing
from loguru import logger
from app.database import AsyncSession
from app import models
from sqlalchemy import select
from urllib.parse import urlparse

if typing.TYPE_CHECKING:
    from app.models import Actor as ActorModel

import app.activitypub as ap


class BaseActor:
    def __init__(self, ap_actor: ap.RawObject) -> None:
        if (ap_type := ap_actor.get("type")) not in ap.ACTOR_TYPES:
            raise ValueError(f"Unexpected actor type: {ap_type}")

        self._ap_actor = ap_actor
        self._ap_type : str = ap_type # type: ignore

    @property
    def ap_actor(self) -> ap.RawObject:
        return self._ap_actor

    @property
    def inbox_url(self) -> str:
        return self.ap_actor["inbox"]

    @property
    def ap_type(self) -> str:
        return self._ap_type


async def fetch_actor(
        db_session : AsyncSession,
        actor_id : str,
) -> "ActorModel":
    exist_actor = (
        await db_session.scalars(
            select(models.Actor).where(
                models.Actor.ap_id == actor_id
            )
        )
    ).one_or_none()

    if not exist_actor:
        ap_object = await ap.fetch(actor_id)
        exist_actor = await save_actor(ap_object, db_session)
        return exist_actor

    return exist_actor

async def save_actor(
        ap_object : dict,
        db_session : AsyncSession
) -> "ActorModel":
    logger.info("save actor " + ap_object["id"])
    actor = models.Actor(
        ap_id=ap_object["id"],
        ap_actor=ap_object,
        ap_type=ap_object["type"],
        handle=_handle(ap_object),
    )

    db_session.add(actor)
    await db_session.flush()
    await db_session.refresh(actor)
    return actor

def _handle (
    ap_object :dict
) -> str:
    ap_id = urlparse(ap_object["id"])
    if not ap_id.hostname:
        raise ValueError(f"Invalid actor ID {ap_id}")

    handle = '@' + ap_object["preferredUsername"] + '@' + ap_id.hostname

    return handle