medium_celery_task_retry.py

from __future__ import absolute_import, unicode_literals

import os
import json
import requests

from celery.task import task
from pushsafer import Client, init


init(os.getenv('PUSHSAFER_PRIVATE_KEY'))
client = Client("")


@task(name='latest_bitcoin_price_notify',
      autoretry_for=(Exception,),
      exponential_backoff=2, retry_kwargs={'max_retries': 5},
      retry_jitter=False)
def latest_bitcoin_price_notify(currency_code: str = None) -> object:
    """
    CoinDesk Bitcoin Price Index (XBP) api fetches bitcoin prices that represents
    an average of bitcoin prices across leading global exchanges
    broadcasts price to external devices via pushsafer API
    Function consists of exponential_backoff backs of gradually with a max of 5 retries
    in the event of an exception of type Exception
    :type currency_code: str
    :param currency_code: coinDesk supported currency
    """
    bpi_url = 'https://api.coindesk.com/v1/bpi/currentprice.json'

    if (currency_code and isinstance(currency_code, str) and currency_code.upper() in ['USD', 'GBP', 'EUR', 'CNY']):
        bpi_url = 'https://api.coindesk.com/v1/bpi/currentprice/{}.json'.format(currency_code.upper())

    response = requests.get(bpi_url)

    if response.status_code != 200:
        raise Exception(f'GET {bpi_url} returned unexpected response code: {response.status_code}')

    data = json.loads(response.content.decode('utf-8'))

    price_data = 'Latest Bitcoin Price {} {}'.format(currency_code, data.get('bpi').get(currency_code).get('rate_float'))

    client.send_message(price_data, "Bitcoin Price",
                        "24745", "1", "50", "2",
                        "https://www.pushsafer.com", "Open Pushsafer",
                        "0", "1", "120", "1200", "0", "", "", "")