Building Python Package: API Client for YouTube Channel Details (RapidAPI)

Hi!
Today, we're going to work on creating a Python 3 package to facilitate access to an API for fetching YouTube channel details (The Better YouTube Channel Details). Even if the topic of API communication doesn't interest you, you might enjoy the presented complete process of creation.

The entire content will be divided into three separate posts, as I don't want to clutter your time with one large monolith.

1. 👉 we are here: creating the package structure and writing code to communicate with the API
2. testing the solution
3. publishing our work on pypi.org

Don't frown! I invite you on an exciting adventure from which you'll surely learn something.

Everyone ready? Camera! ACTION! 🎬

🧬 The Project Structure

We're creating the structure of our package in Python 3

Every package must adhere to a specific directory and file structure. According to the official documentation.

packaging_tutorial/
├── LICENSE
├── pyproject.toml
├── README.md
├── src/
│   └── example_package_YOUR_USERNAME_HERE/
│       ├── __init__.py
│       └── example.py
└── tests/

This is how the package should look.

Our API will be relatively very simple:

• we will only connect to the API
• fetch JSON data
• map JSON to an object so that it can be easily used in other applications

Let's create the file structure of our project.

🧙 Doing some magic

Looks great, so let's move on to writing the code, and we'll handle the metadata files later. As programmers, we love writing code, so anything not related to code isn't as appealing to us 😈

⌨️ "Pressing (random) keys"

So, let's program! That's what we're paid for (right...? right!?)

We're adding a file src/api_client.py. This file will contain all the communication with the API and the mapping of the response. The constructor of the class should accept a rapid_api_key, which we can obtain. How to do it? It's simple:

✔️ register (for free of course)
✔️ move to the API page and click on Test Endpoint
✔️ on the same page, grab the displayed value of X-RapidAPI-Key, this is our rapid_api_key

Great, let's write the code:

class YouTubeChannelDetailsApiClient(object):
    def __init__(self, rapid_api_key: str):
        self.rapidapi_host = 'the-better-youtube-channel-details.p.rapidapi.com'
        self.headers = {
            "X-RapidAPI-Key": rapid_api_key,
            "X-RapidAPI-Host": self.rapidapi_host,
        }

Our structure has changed

Looks decent. There's nothing particularly interesting in it 🥱 a bit dull.

We need a function that will query the API. We'll use the aiohttp package for assistance. This package provides us with an HTTP client for communicating with the API.

✋ Before we proceed, we need to install this package:
pip install aiohttp
so we can create HTTP requests.

Our function for communicating with the API will be named __get_request. From the name, it's clear that it will be a private function accessible only within api_client.

The function should accept path, which is the path to the API endpoint, and query. It's important to note that query should be of type dict. Why dict and not, string? We want to encode each parameter input by the user, and we'll use the urlencode function for this, which accepts a dict, not a string.

Okay, but what should the __get_request function return? The natural instinct is a string, so we'll set it to return a string (wow, what a surprise! 😉)

Let's write the header of the __get_request function:

async def __get_request(self, path: str, query: dict) -> str:

Moving on to programming, which is what we love the most! ;) the __get_request function should:

• build the URL to the API endpoint
• query the API
• in case of a network error, retry the request after waiting 300 milliseconds
• in case of an API error, throw an exception

Let's start writing the code for this function. Remember, we're using aiohttp for the HTTP client, and we'll need to handle retries and error checking.

    async def __get_request(self, path: str, query: dict) -> str:
        url = URL(
            host=self.rapidapi_host,
            path=path,
            query=query,
            scheme="https"
        ).as_string()

        session_timeout = aiohttp.ClientTimeout(total=45.0)
        is_ok = False

        while True:
            async with aiohttp.ClientSession(headers=self.headers, timeout=session_timeout) as session:
                async with session.get(url=url) as response:
                    try:
                        json_response = await response.json()  
                        if response.status == 200:  
                            is_ok = True
                    except aiohttp.client.ContentTypeError:
                        continue
                    if is_ok:
                        break
            await asyncio.sleep(0.3)

        return json_response

Let's check to make sure we have all the necessary imports added:

import asyncio
import aiohttp

from urllib.parse import urlencode
from purl import URL

Sure, let's also ensure we have all the necessary packages installed:
✔️ pip install asyncio
✔️ pip install aiohttp
✔️ pip install purl

The code looks good (or even great!), but does it work well? That's the eternal question we'll only answer by testing it ourselves.

But that will come in the next post, so stay tuned!

I hope you enjoyed the style and examples. If possible, please give a thumbs up, comment, or any kind of reaction! It fuels me for future posts. Follow me to make sure you don't miss anything 🚀

Thanks for your time and see you in the next post!