microblog.pub/tasks.py

import os
import json
import logging
import random

import requests
from celery import Celery
from requests.exceptions import HTTPError

from config import HEADERS
from config import ID
from config import DB
from config import KEY
from config import USER_AGENT
from utils.httpsig import HTTPSigAuth
from utils.opengraph import fetch_og_metadata
from utils.linked_data_sig import generate_signature


log = logging.getLogger(__name__)
app = Celery('tasks', broker=os.getenv('MICROBLOGPUB_AMQP_BROKER', 'pyamqp://guest@localhost//'))
SigAuth = HTTPSigAuth(ID+'#main-key', KEY.privkey)


@app.task(bind=True, max_retries=12)
def post_to_inbox(self, payload: str, to: str) -> None:
    try:
        log.info('payload=%s', payload)
        log.info('generating sig')
        signed_payload = json.loads(payload)
        generate_signature(signed_payload, KEY.privkey)
        log.info('to=%s', to)
        resp = requests.post(to, data=json.dumps(signed_payload), auth=SigAuth, headers={
            'Content-Type': HEADERS[1],
            'Accept': HEADERS[1],
            'User-Agent': USER_AGENT,    
        })
        log.info('resp=%s', resp)
        log.info('resp_body=%s', resp.text)
        resp.raise_for_status()
    except HTTPError as err:
        log.exception('request failed')
        if 400 >= err.response.status_code >= 499:
            log.info('client error, no retry')
            return
        self.retry(exc=err, countdown=int(random.uniform(2, 4) ** self.request.retries))


@app.task(bind=True, max_retries=12)
def fetch_og(self, col, remote_id):
    try:
        log.info('fetch_og_meta remote_id=%s col=%s', remote_id, col)
        if col == 'INBOX':
            log.info('%d links saved', fetch_og_metadata(USER_AGENT, DB.inbox, remote_id))
        elif col == 'OUTBOX':
            log.info('%d links saved', fetch_og_metadata(USER_AGENT, DB.outbox, remote_id))
    except Exception as err:
        self.log.exception('failed')
        self.retry(exc=err, countdown=int(random.uniform(2, 4) ** self.request.retries))