Write a Class for Calling APIs Based on Python

Nowadays, back - end development mainly involves writing various APIs for others to use. In my daily work, I not only write APIs but also frequently call APIs written by others.

Let me share the module for calling APIs that I often use.

Before looking at the code, there are some assumptions that can help you understand the code.

Some Assumptions

Suppose we have an API: http://127.0.0.1:8000/api/token. For detailed information, you can refer to simple jwt.

Here, I’ll provide a simple interface document as follows.

Request Method

POST

Request Parameters

The following data in JSON format needs to be provided in the request body:

• username: Username
• password: Password

Example:

{     "username": "<username>",     "password": "<password>" }

Response Content

If the authentication is successful, the interface will return a response in the following format:

{     "access": "<Access_Token>",     "refresh": "<Refresh_Token>" }

Among them:

• <Access_Token>: Access token, which can be used for subsequent protected operations.
• <Refresh_Token>: Refresh token, which can be used to obtain a new access token after the access token expires.

Error Handling

• If the username or password is incorrect, a 401 Unauthorized error will be returned, along with a descriptive error message.
• If an internal server error occurs, a 500 Internal Server Error will be returned.

Usage Example

Request

curl --header "Content-Type: application/json" \   --request POST \   --data '{"username":"admin", "password":"123456"}' \   http://localhost:8000/token/

Response

{     "access": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6Ikpv...",     "refresh": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6Ikpv..." }

APIConnection

With the above assumptions, we can now look at the code.

The following is the entire code, and the explanations are placed in the comments.

import json
import logging
import os
import requests


logging.basicConfig(
    format="%(asctime)s %(levelname)-8s %(message)s",
    level=logging.INFO,
    datefmt="%Y-%m-%d %H:%M:%S",
)


class APIConnection:
    """
    Api Connection
    """

    def __init__(self):
        # Get the backend host from the environment variable instead of hard - coding
        self.api_url = os.environ.get("BACKEND_API_URL", "http://localhost:8000/")

        self.token = ""
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": "Bearer {}".format(self.token),
            "Cache-Control": "no-cache",
        }

    def request_jwt(self):
        """
        Used to call /api/token to obtain a token.
        When calling, the username and password need to be obtained from the environment variables first.
        After obtaining the token, update the self.headers attribute for authentication when making subsequent requests.
        """
        self.headers = {
            "Content-Type": "application/json",
            "Cache-Control": "no-cache",
        }

        api_url = f"{self.api_url}api/token/"

        data = {
            "username": os.environ["SCRIPT_USER"],
            "password": os.environ["SCRIPT_PASSWORD"],
        }

        res = requests.post(
            api_url, data=json.dumps(data), headers=self.headers, timeout=60
        )

        if res.status_code == 200:
            data = json.loads(res.text)

            self.token = data["access"]

            self.headers = {
                "Content-Type": "application/json",
                "Authorization": "Bearer {}".format(self.token),
                "Cache-Control": "no-cache",
            }

            return self.token

        logging.info("Invalid.", res.text)
        return False

    def get_data(self, path, limit=100):
        """
        Retrieve data from a specified URI (i.e., the path parameter) and return it to the front - end.
        By default, it's 100 pieces of data,
        applicable in cases where no filtering is done.
        """

        self.request_jwt()

        request_url = f"{self.api_url}{path}?limit={limit}"
        response = requests.get(request_url, headers=self.headers, timeout=60)

        if response.status_code == 200:
            data = response.json()

            return data

        logging.error(
            "statue code: {}. error message: {}".format(
                response.status_code, response.text
            )
        )

        return None

    def post_data(self, item, api):
        """
        Create a new piece of data
        """
        self.request_jwt()
        api_url = f"{self.api_url}{api}"

        try:
            response = requests.post(
                api_url, data=json.dumps(item), headers=self.headers, timeout=60
            )

            if response.status_code == 403:
                if self.request_jwt():
                    self.post_data_with_response(item, api)
                else:
                    logging.error("Invalid Credentials")

            elif response.status_code not in [200, 201]:
                logging.error(response.text)
                logging.error(f"{response.status_code} - unable to post data")

        except Exception as err:
            logging.error(err)

    def update_data(self, api, item):
        """
        Update data.
        Different from creating new data, this is to update existing data. Note to pass in the primary key.
        """

        api_url = f"{self.api_url}{api}"
        self.request_jwt()
        try:
            response = requests.patch(
                url=api_url, data=json.dumps(item), headers=self.headers, timeout=60
            )

            if response.status_code == 403:
                # try again
                self.update_data(api, item)

            elif response.status_code not in [200, 201]:
                logging.info(response.status_code)

        except Exception as err:
            logging.error(err)
            return err

        return response

When writing this blog, I found that this code is really not that good. There are the following problems:

1. The exception handling is only through logging.
2. Each method requests self.request_jwt() separately, which not only puts unnecessary pressure on the backend but also increases its own time consumption.
3. request.Session() can be used to maintain some header parameters and utilize the connection pool to improve performance.

This is probably the significance of writing a blog. When writing, you are actually doing a review.