自学内容网 自学内容网
自学内容网 > 正文

restful-python

🕗 发布于 2024-11-13 03:48 restful  python  后端  
#encoding:utf-8
import requests,json,time
def get_auth_token(controller_ip,domain,name,password):
    try:
        url = f"http://{controller_ip}:5000/v3/auth/tokens"
        body = {
            "auth": { 
                "identity": {
                    "methods": ['password'],
                    "password": {
                        "user": {
                            "domain": {"name": domain},
                            "name": name,
                            "password": password,
                        }
                    }
                },
                "scope": {
                    "project": {
                        "domain": {"name": domain},
                        "name": name
                    }
                }
            }
        }
        headers = {
            "Content-Type": "application/json"
        }
        token = requests.post(url,headers=headers,data=json.dumps(body)).headers['X-Subject-Token']
        headers = {
            "X-Auth-Token": token
        }
        print(f"token值为:{token}")
        return headers
    except Exception as e:
        print(f"token获取失败,{e}")

class image_manager:
    def __init__(self,handers:dict,resUrl):
        self.headers = handers
        self.resUrl = resUrl

    def create_image(self,image_name,disk_format,container_format):
        body = {
            "name": image_name,
            "disk_format": disk_format,
            "container_format": container_format,
        }

        req = requests.post(self.resUrl,headers=self.headers,data=json.dumps(body)).text
        print(f"创建镜像的信息为:{req}")
        return req

    def get_image_id(self,name):
        req = json.loads(requests.get(self.resUrl,headers=self.headers).text)
        for image in req['images']:
            if image['name'] == name:
                return image['id']
        return "NONE"

    def upload_image(self,id,file_path:str):
        url = self.resUrl + "/" + id + "/file"
        self.headers["Content-Type"] = "application/octet-stream"
        req = requests.put(url,headers=self.headers,data=open(r'C:\Users\admin\Desktop\cirros-0.5.1-x86_64-disk.img','rb').read())
        if req.status_code == 204:
            print("上传镜像成功",req.status_code)
        else:
            print("上传镜像失败",req.status_code)

        print(f"镜像上传信息:{req}")
        return req

    def get_image(self,id):
        url = self.resUrl + "/" + id
        req = json.loads(requests.get(self.resUrl,headers=self.headers).text)
        print(f"获取到的镜像信息为:{req}")
        return req

    def delete_image(self,id):
        url = self.resUrl + "/" + id
        req = requests.delete(url,headers=self.headers)
        print(f"删除信息:{req}")
        return req

if __name__ == "__main__":
    controller_ip = "192.168.200.10"
    domain = "demo"
    name = "admin"
    password = "123"
    headers = get_auth_token(controller_ip, domain, name, password)
    image_m = image_manager(headers,f"http://{controller_ip}:9292/v2/images")

    #create
    create_image = image_m.create_image("cirros001","qcow2","bare")

    #get id
    get_id = image_m.get_image_id("cirros001")
    print(f"cirros001镜像ID为:{get_id}")

    #upload
    upload_image = image_m.upload_image(get_id,"cirros-0.3.4-x86_64-disk.img")

    #get image
    get_image = image_m.get_image(get_id)
    with open("image_demo.json","w")as outfile:
        json.dump(get_image,outfile,indent=4)

这个代码的作用是通过 REST API 接口和 OpenStack 的镜像服务（Glance）进行交互，实现创建、查询、上传、删除镜像等操作。下面我们来逐步详细解读每个部分的功能和流程：

1. 导入必要的库

import requests, json, time
  • requests: 用于发送 HTTP 请求的库，这里主要用于与 OpenStack 的 API 通信。
  • json: 用于处理 JSON 格式的数据，这是 OpenStack API 使用的数据格式。
  • time: 用于延迟处理或等待，但在此代码中没有被实际使用。

2. 获取认证 Token 函数 get_auth_token

def get_auth_token(controller_ip, domain, name, password):
    try:
        url = f"http://{controller_ip}:5000/v3/auth/tokens"
        body = {
            "auth": {
                "identity": {
                    "methods": ["password"],
                    "password": {
                        "user": {
                            "domain": {"name": domain},
                            "name": name,
                            "password": password,
                        }
                    }
                },
                "scope": {
                    "project": {
                        "domain": {"name": domain},
                        "name": name
                    }
                }
            }
        }
        headers = {"Content-Type": "application/json"}
        token = requests.post(url, headers=headers, data=json.dumps(body)).headers['X-Subject-Token']
        headers = {"X-Auth-Token": token}
        print(f"token值为:{token}")
        return headers
    except Exception as e:
        print(f"token获取失败,{e}")
功能
  • 参数说明：
    • controller_ip: 控制节点的 IP 地址。
    • domain: 用户域名。
    • name: 用户名。
    • password: 用户密码。
  • 流程：
    1. 构造 URL 指向身份认证服务的端点。
    2. 定义认证的请求主体 body，格式符合 OpenStack 认证 API 需要的 JSON 格式。
    3. 通过 POST 请求发送给身份认证服务，并从响应头中获取 X-Subject-Token 值，作为后续操作的认证 Token。
    4. 返回 headers 包含认证 Token，供后续函数调用使用。
  • 异常处理：如果发生错误，打印错误信息。

3. 定义镜像管理类 image_manager

这个类包括对镜像的创建、获取、上传和删除等操作。

3.1 初始化方法 __init__
class image_manager:
    def __init__(self, headers: dict, resUrl):
        self.headers = headers
        self.resUrl = resUrl
  • 功能：类的初始化方法，用于设置请求头（含认证 Token）和镜像服务的基本 URL。
  • 参数：
    • headers: 认证头部信息。
    • resUrl: OpenStack 镜像服务的 API 端点 URL。
3.2 创建镜像方法 create_image
def create_image(self, image_name, disk_format, container_format):
    body = {
        "name": image_name,
        "disk_format": disk_format,
        "container_format": container_format,
    }
    req = requests.post(self.resUrl, headers=self.headers, data=json.dumps(body)).text
    print(f"创建镜像的信息为:{req}")
    return req
  • 功能：向镜像服务发送请求创建新镜像。
  • 参数：
    • image_name: 镜像名称。
    • disk_format: 磁盘格式，如 qcow2。
    • container_format: 容器格式，如 bare。
  • 流程：将镜像信息发送到 self.resUrl，获取并打印返回的创建结果。
3.3 获取镜像 ID 方法 get_image_id
def get_image_id(self, name):
    req = json.loads(requests.get(self.resUrl, headers=self.headers).text)
    for image in req['images']:
        if image['name'] == name:
            return image['id']
    return "NONE"
  • 功能：查询指定名称的镜像 ID。
  • 参数：name 要查询的镜像名称。
  • 流程：遍历获取的镜像信息列表，如果找到匹配的镜像名称，则返回该镜像的 id，否则返回 "NONE"。
3.4 上传镜像文件方法 upload_image
def upload_image(self, id, file_path: str):
    url = self.resUrl + "/" + id + "/file"
    self.headers["Content-Type"] = "application/octet-stream"
    req = requests.put(url, headers=self.headers, data=open(r'C:\Users\admin\Desktop\cirros-0.5.1-x86_64-disk.img', 'rb').read())
    if req.status_code == 204:
        print("上传镜像成功", req.status_code)
    else:
        print("上传镜像失败", req.status_code)
    print(f"镜像上传信息:{req}")
    return req
  • 功能：向镜像服务上传指定镜像文件。
  • 参数：
    • id: 镜像 ID。
    • file_path: 镜像文件路径。
  • 流程：使用 PUT 请求上传镜像文件，检查是否上传成功并返回结果。
3.5 获取镜像信息方法 get_image
def get_image(self, id):
    url = self.resUrl + "/" + id
    req = json.loads(requests.get(self.resUrl, headers=self.headers).text)
    print(f"获取到的镜像信息为:{req}")
    return req
  • 功能：根据镜像 ID 获取详细信息。
  • 参数：id 镜像 ID。
  • 流程：获取镜像的详细信息，并将信息打印输出。
3.6 删除镜像方法 delete_image
def delete_image(self, id):
    url = self.resUrl + "/" + id
    req = requests.delete(url, headers=self.headers)
    print(f"删除信息:{req}")
    return req
  • 功能：删除指定的镜像。
  • 参数：id 镜像 ID。
  • 流程：根据 ID 发送删除请求，删除镜像。

4. 主函数部分

if __name__ == "__main__":
    controller_ip = "192.168.200.10"
    domain = "demo"
    name = "admin"
    password = "123"
    headers = get_auth_token(controller_ip, domain, name, password)
    image_m = image_manager(headers, f"http://{controller_ip}:9292/v2/images")

    # create
    create_image = image_m.create_image("cirros001", "qcow2", "bare")

    # get id
    get_id = image_m.get_image_id("cirros001")
    print(f"cirros001镜像ID为:{get_id}")

    # upload
    upload_image = image_m.upload_image(get_id, "cirros-0.3.4-x86_64-disk.img")

    # get image
    get_image = image_m.get_image(get_id)
    with open("image_demo.json", "w") as outfile:
        json.dump(get_image, outfile, indent=4)
  • 流程：
    1. 通过 get_auth_token 函数获取认证 Token。
    2. 创建镜像管理对象 image_m。
    3. 调用 create_image 创建名为 “cirros001” 的镜像。
    4. 获取该镜像的 ID。
    5. 上传镜像文件。
    6. 获取镜像信息，并将信息保存到 image_demo.json 文件。

个人疑惑

为什么body里面的都要用"" ?

在这个 body 字典中，用双引号 "" 包裹的部分代表字符串字面量，而没有使用引号的部分是变量。Python 中，字符串字面量和变量的使用有不同的规则：

对的，正是因为这些字段是字符串才需要加引号。在 Python 中，字符串需要用引号（单引号 ' ' 或双引号 " "）包裹，这样 Python 解释器才能识别它们是字符串字面量，而不是变量或其他类型的值。

在 body 字典里，加引号的部分（如 "auth"、"identity"）是固定的字符串字段名，代表 OpenStack API 请求中要求的键值；而没有引号的部分（如 domain、name）是变量名，它们在代码运行时会被替换成变量的值。

所以，总结来说，因为字符串是固定的文本信息，所以在代码中需要加引号来让 Python 把它识别为字符串，而不是解释成变量名或其他内容。

“methods”: [“password”],为什么这里要用方括号?

在 methods 字段里使用方括号 [] 是因为它的值是一个列表（数组），而不是单一字符串。在 Python 中，列表是一种数据类型，可以包含多个元素。即使在这个例子里列表中只有一个元素（"password"），也需要使用方括号将其包裹起来，以表明这个字段是一个列表类型，而不是一个单独的字符串。

methods 字段的设计确实是为了支持多个认证方式，所以它必须是一个列表。虽然在这个例子里，methods 里面只有一个 "password" 方法，但如果需要的话，完全可以向这个列表里添加更多的认证方式，比如 "token" 等。

这段代码是用于与 OpenStack 镜像服务（Glance）进行交互的脚本，具体执行了创建、查询、上传和删除镜像的操作。我们可以逐步解析它们的功能和作用。

代码结构与流程

1. 定义 image_manager 类

这个类主要封装了与 OpenStack 镜像服务进行交互的相关操作。它包括以下方法：

  • __init__：构造函数，初始化了类的两个成员变量：
    • self.headers: 存储用于 HTTP 请求的头信息（包含认证 Token）。
    • self.resUrl: 存储镜像服务的 URL。
2. create_image 方法

用于创建一个新的镜像。

  • 接受镜像的 名称、磁盘格式 和 容器格式 作为参数。
  • 使用 requests.post 方法发送一个 HTTP POST 请求到 self.resUrl，并传递镜像的相关信息（JSON 格式）。请求成功后，返回镜像创建的响应信息。
3. get_image_id 方法

用于通过镜像的名称查询镜像的 ID。

  • 发送一个 GET 请求到 self.resUrl，获取所有镜像的信息。
  • 遍历返回的镜像列表，通过镜像名称匹配相应的镜像，返回其 ID。如果没有找到匹配的镜像，则返回 "NONE"。
4. upload_image 方法

用于上传镜像文件。

  • 通过镜像的 ID 和 文件路径 来上传镜像文件。请求是一个 PUT 请求，将镜像文件以二进制流上传到 OpenStack 镜像服务的指定 URL。
  • 如果上传成功（HTTP 状态码为 204），则打印成功信息；否则打印失败信息。
5. get_image 方法

用于获取指定 ID 的镜像信息。

  • 发送一个 GET 请求到 self.resUrl 来获取指定 ID 的镜像详情，并打印响应信息。
6. delete_image 方法

用于删除指定 ID 的镜像。

  • 发送一个 DELETE 请求到指定镜像的 URL，删除该镜像。
7. if __name__ == "__main__"

这部分代码用于执行类方法，具体做了以下操作：

  • 获取认证 Token：

    • 通过调用 get_auth_token 函数获取认证 Token，并作为请求头 headers 提供给 image_manager 类。get_auth_token 函数的作用是通过 OpenStack 的身份验证接口获取认证 Token，该 Token 在后续的 API 请求中作为认证凭证使用。

  • 创建镜像：

    • 调用 create_image 方法，创建一个名称为 cirros001，磁盘格式为 qcow2，容器格式为 bare 的镜像。

  • 获取镜像 ID：

    • 调用 get_image_id 方法，通过镜像名称 cirros001 查询镜像的 ID。

  • 上传镜像文件：

    • 调用 upload_image 方法，将本地的镜像文件 cirros-0.3.4-x86_64-disk.img 上传到 OpenStack 镜像服务。上传的是上一步通过 get_image_id 获取的镜像 ID。

  • 获取镜像信息：

    • 调用 get_image 方法，通过镜像 ID 获取镜像的详细信息，并将其保存到本地文件 image_demo.json 中。

每个函数的详细解释

  1. get_auth_token：

    • 获取身份验证 Token，用于后续 API 调用的认证。
    • 它通过向 OpenStack 的身份验证端点发送请求，返回一个 Token。这个 Token 会在请求头中传递，确保后续 API 调用时能正确识别身份。

  2. create_image：

    • 使用 POST 请求向镜像服务创建一个新镜像。请求体包含镜像的名称、磁盘格式和容器格式。

  3. get_image_id：

    • 使用 GET 请求获取镜像列表，遍历每个镜像，并返回匹配名称的镜像 ID。

  4. upload_image：

    • 使用 PUT 请求上传镜像文件。请求体包含镜像文件的二进制数据。

  5. get_image：

    • 使用 GET 请求获取指定 ID 镜像的详细信息，并打印响应数据。

  6. delete_image：

    • 使用 DELETE 请求删除指定 ID 的镜像。

整体流程

  1. 认证阶段：首先，调用 get_auth_token 获取身份验证 Token。
  2. 镜像创建：然后调用 create_image 创建一个新镜像。
  3. 获取镜像 ID：接着，使用 get_image_id 获取该镜像的 ID。
  4. 镜像文件上传：通过调用 upload_image 上传镜像文件。
  5. 获取镜像信息：最后，通过 get_image 获取镜像的详细信息，并保存为 JSON 文件。

总结

这段代码通过封装类 image_manager 和其中的几个方法，提供了与 OpenStack 镜像服务（Glance）进行交互的能力。它实现了镜像的创建、查询、上传和删除等常见操作，适用于需要自动化管理 OpenStack 镜像的场景。


原文地址：https://blog.csdn.net/m0_74752717/article/details/143725601

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

相关文章

自学内容网
Copyright © 2015-2024