基于Python编写一个调用API的类

panzhixiang

发布于 2024-10-30 19:41:30

3790

文章被收录于专栏：panzhixiangpanzhixiang

现在后端开发基本上都是写各种API提供给别人使用，我在日常工作里既写API，也经常调用别人写的API。

分享一下经常使用的调用API的模块。

看代码之前会有一些假设，可以帮助理解代码。

一些假设

假设我们有一个API是： http://127.0.0.1:8000/api/token ，这个详细信息可以参考simple jwt 。

我在这里给一个简单的接口文档，如下。

请求方法

POST

请求参数

在请求体中需要提供以下json格式的数据：

username: 用户名
password: 密码

示例：

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

响应内容

如果认证成功，接口将返回如下格式的响应：

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

其中：

<Access_Token>: 访问令牌，可以用来进行后续的受保护操作。
<Refresh_Token>: 刷新令牌，可以用来在访问令牌过期后获取新的访问令牌。

错误处理

如果用户名或密码错误，会返回401 Unauthorized错误，并且具有描述性的错误信息。
如果服务器内部发生错误，会返回500 Internal Server Error错误。

使用示例

请求

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

响应

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

APIConnection

有了上面的假设，现在就可以来看代码了。

代码如下，直接贴出了全部代码，需要解释的内容放在了注释中。

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):
        # 通过环境变量来获取后端的host，而不是硬编码
        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):
        """
        用于调用/api/token获取token，
        调用的时候需要先从环境变量中获取用户名和密码，
        获得token之后更新self.headers属性，便于后面发起请求的时候做认证        
        """
        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):
        """
        从一个指定的uri（也就是path参数）获取数据返回给前端，
        默认是100条数据，
        使用于不做任何过滤的情况
        """

        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):
        """
        新建一条数据
        """
        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):
        """
        更新数据，
        区别于新建数据，是对已有的数据进行更新，要注意传入主键
        """

        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

写这篇博客的时候发现这段代码写的真心不咋地，有以下几个问题