bitly_shorten_async

bitly_shorten_app.py

# Non-blocking script for shorten link via https://bit.ly Api

import asyncio
import logging
from typing import Dict, Optional

from aiohttp import ClientSession


logger = logging.getLogger('bitly_app')
logger.setLevel(logging.DEBUG)
# create console handler with a higher log level
ch = logging.StreamHandler()
formatter = logging.Formatter(
    (
        '{"unix_time":%(created)s, "time":"%(asctime)s", "module":"%(name)s",'
        ' "line_no":%(lineno)s, "level":"%(levelname)s", "msg":"%(message)s"},'
    )
)
ch.setFormatter(formatter)
logger.addHandler(ch)

# Get your token: https://support.bitly.com/hc/en-us/articles/230647907-How-do-I-generate-an-OAuth-access-token-for-the-Bitly-API-
TOKEN = "" 
URL = 'https://api-ssl.bitly.com/v4/'
oauth = {'Authorization': f'Bearer {TOKEN}'}


async def fetch(session: ClientSession,
                headers: Dict[str, str],
                url: str,
                payload: Dict,
                endpoint: str,
                method='GET',
                ) -> Optional[Dict]:
    """Generic get/post request"""
    async with session.request(method=method,
                               url=f'{url}{endpoint}',
                               headers=headers,
                               json=payload) as response:

        if response.status in (200, 201):
            json_response = await response.json()
            logger.debug(json_response)
            return json_response
        else:
            logger.error(f"Failed to make request {response.status}")

    return None


async def shorten(link: str) -> Optional[str]:
    """https://dev.bitly.com/v4_documentation.html#operation/createBitlink"""
    session = ClientSession()
    result = None
    payload = {
        "domain": "bit.ly",
        "long_url": link
    }
    try:
        result = await fetch(session, oauth, URL, payload, 'shorten', method='POST')
    except Exception as exc:
        logger.error(exc)
    finally:
        await session.close()
        return result.get('link', None) if result else None

if __name__ == '__main__':
    example_link = 'https://dev.bitly.com'
    loop = asyncio.get_event_loop()
    loop.run_until_complete(shorten(example_link))