microblog.pub/little_boxes/remote_object.py

import logging
from typing import Any

import requests

from .urlutils import check_url
from .errors import ActivityNotFoundError

logger = logging.getLogger(__name__)


class DefaultRemoteObjectFetcher(object):
    """Not meant to be used on production, a caching layer, and DB shortcut fox inbox/outbox should be hooked."""

    def __init__(self):
        self._user_agent = 'Little Boxes (+https://github.com/tsileo/little_boxes)'

    def fetch(self, iri):
        check_url(iri)

        resp = requests.get(iri, headers={
            'Accept': 'application/activity+json',
            'User-Agent': self._user_agent,
        })

        if resp.status_code == 404:
            raise ActivityNotFoundError(f'{iri} cannot be fetched, 404 not found error')

        resp.raise_for_status()

        return resp.json()


OBJECT_FETCHER = DefaultRemoteObjectFetcher()


def set_object_fetcher(object_fetcher: Any):
    global OBJECT_FETCHER
    OBJECT_FETCHER = object_fetcher