1457 字
7 分钟
OpenStack平台Python运维开发
2025-03-20
系统运维
OpenStack

OpenStack 平台 Python 运维开发#

API 端点速查#

服务默认端口API 路径用途
Keystone5000/v3/auth/tokens, /v3/users认证、用户管理
Glance9292/v2/images, /v2/images/{id}/file镜像管理
Nova8774/v2.1/flavors, /v2.1/servers云主机、规格
Cinder8776/v2/{project}/volumes云硬盘
Neutron9696/v2.0/networks, /v2.0/subnets网络

两种开发方式：RESTful API（requests 库，灵活可控）和 Python SDK（openstacksdk，代码更简洁）。

1. Keystone 认证模块#

所有 RESTful API 调用都需要先获取 Token。以下函数返回带 Token 的 Header。

import json, requests




def get_auth_token(controller_ip, domain, user, password):
    """获取 Keystone Token，返回带认证信息的 Header"""
    url = f"http://{controller_ip}:5000/v3/auth/tokens"
    body = {
        "auth": {
            "identity": {
                "methods": ["password"],
                "password": {
                    "user": {
                        "domain": {"name": domain},
                        "name": user,
                        "password": password
                    }
                }
            },
            "scope": {
                "project": {
                    "domain": {"name": domain},
                    "name": user
                }
            }
        }
    }
    headers = {"Content-Type": "application/json"}
    resp = requests.post(url, data=json.dumps(body), headers=headers)
    token = resp.headers['X-Subject-Token']
    return {"X-Auth-Token": token, "Content-Type": "application/json"}




# 使用示例
headers = get_auth_token("192.168.1.10", "demo", "admin", "000000")

2. 用户管理（RESTful API）#

操作 Keystone /v3/users 端点，实现用户的增删改查。

import json, requests




class KeystoneUserManager:
    def __init__(self, headers: dict, controller_ip: str):
        self.headers = headers
        self.url = f"http://{controller_ip}:5000/v3/users"


    def create_user(self, name, password, description=""):
        """创建用户"""
        body = {"user": {"name": name, "password": password, "description": description}}
        resp = requests.post(self.url, data=json.dumps(body), headers=self.headers)
        return resp.text


    def list_users(self):
        """查询所有用户"""
        return requests.get(self.url, headers=self.headers).text


    def get_user_id(self, name):
        """根据名称查找用户 ID"""
        result = json.loads(self.list_users())
        for user in result['users']:
            if user['name'] == name:
                return user['id']
        return None


    def get_user(self, user_id):
        """根据 ID 查询用户详情"""
        resp = requests.get(f"{self.url}/{user_id}", headers=self.headers)
        return json.loads(resp.text)


    def delete_user(self, name):
        """根据名称删除用户"""
        user_id = self.get_user_id(name)
        if not user_id:
            return {"error": "User not found"}
        resp = requests.delete(f"{self.url}/{user_id}", headers=self.headers)
        if resp.status_code == 204:
            return {"UserDeletedSuccess": True}
        return json.loads(resp.text)


    def update_password(self, user_id, original_password, new_password):
        """修改用户密码"""
        body = {"user": {"password": new_password, "original_password": original_password}}
        resp = requests.post(f"{self.url}/{user_id}/password",
                             data=json.dumps(body), headers=self.headers)
        if resp.status_code == 204:
            return {"PasswordUpdateSuccess": True}
        return json.loads(resp.text)

3. 镜像管理（RESTful API）#

操作 Glance /v2/images 端点。镜像上传分两步：先创建元数据，再 PUT 二进制数据。

import json, requests




class GlanceImageManager:
    def __init__(self, headers: dict, controller_ip: str):
        self.headers = headers.copy()
        self.url = f"http://{controller_ip}:9292/v2/images"


    def create_image(self, name, container_format="bare", disk_format="qcow2"):
        """创建镜像（仅元数据）"""
        body = {
            "container_format": container_format,
            "disk_format": disk_format,
            "name": name
        }
        resp = requests.post(self.url, data=json.dumps(body), headers=self.headers)
        if resp.status_code == 201:
            return json.loads(resp.text)  # 返回镜像信息（含 id）
        return {"error": resp.text}


    def get_image_id(self, name):
        """根据名称查找镜像 ID"""
        result = json.loads(requests.get(self.url, headers=self.headers).text)
        for img in result['images']:
            if img['name'] == name:
                return img['id']
        return None


    def upload_image_data(self, image_id, file_path):
        """上传镜像二进制数据"""
        upload_url = f"{self.url}/{image_id}/file"
        upload_headers = self.headers.copy()
        upload_headers['Content-Type'] = "application/octet-stream"
        with open(file_path, 'rb') as f:
            resp = requests.put(upload_url, data=f.read(), headers=upload_headers)
        if resp.status_code == 204:
            return {"ImageUploadSuccess": True}
        return {"error": resp.text}


    def list_images(self):
        """查询所有镜像"""
        return requests.get(self.url, headers=self.headers).text


    def get_image(self, image_id):
        """查询单个镜像详情"""
        resp = requests.get(f"{self.url}/{image_id}", headers=self.headers)
        return resp.text


    def delete_image(self, image_id):
        """删除镜像"""
        resp = requests.delete(f"{self.url}/{image_id}", headers=self.headers)
        if resp.status_code == 204:
            return {"ImageDeletedSuccess": True}
        return json.loads(resp.text)

4. Flavor 管理（RESTful API）#

操作 Nova /v2.1/flavors 端点。

import json, requests




class NovaFlavorManager:
    def __init__(self, headers: dict, controller_ip: str):
        self.headers = headers
        self.url = f"http://{controller_ip}:8774/v2.1/flavors"


    def create_flavor(self, name, ram, vcpus, disk, flavor_id):
        """创建 Flavor"""
        body = {"flavor": {
            "name": name, "ram": ram, "vcpus": vcpus,
            "disk": disk, "id": flavor_id
        }}
        resp = requests.post(self.url, data=json.dumps(body), headers=self.headers)
        return resp.text


    def list_flavors(self):
        """查询所有 Flavor"""
        resp = requests.get(self.url, headers=self.headers)
        return json.loads(resp.text)


    def get_flavor(self, flavor_id):
        """查询单个 Flavor"""
        resp = requests.get(f"{self.url}/{flavor_id}", headers=self.headers)
        return json.loads(resp.text)


    def delete_flavor(self, flavor_id):
        """删除 Flavor"""
        resp = requests.delete(f"{self.url}/{flavor_id}", headers=self.headers)
        if resp.status_code == 202:
            return {"FlavorDeletedSuccess": True}
        return json.loads(resp.text)

5. Cinder 卷管理（Python SDK）#

使用 openstacksdk 实现卷创建、扩容、导出为镜像。

import openstack, time




def wait_for_resource_status(conn, resource_type, resource_id, target_status, timeout=120):
    """等待资源达到目标状态"""
    for _ in range(timeout // 2):
        if resource_type == 'volume':
            vol = conn.block_storage.get_volume(resource_id)
            if vol.status == target_status:
                return True
        elif resource_type == 'image':
            img = conn.image.get_image(resource_id)
            if img.status == target_status:
                return True
        time.sleep(2)
    raise TimeoutError(f"{resource_type} {resource_id} 未达到 {target_status}")




def main():
    conn = openstack.connect(
        auth_url='http://<controller>:5000/v3',
        username='admin', password='000000',
        project_name='admin', user_domain_name='demo',
        project_domain_name='demo'
    )


    # 清理同名旧卷
    for vol in conn.block_storage.volumes(name='cinder01'):
        conn.block_storage.delete_volume(vol)


    # 创建 2G 卷
    volume = conn.block_storage.create_volume(
        name='cinder01', size=2, description='Created by Python'
    )
    wait_for_resource_status(conn, 'volume', volume.id, 'available')


    # 扩容到 6G
    conn.block_storage.extend_volume(volume=volume.id, size=6)
    wait_for_resource_status(conn, 'volume', volume.id, 'available')


    # 导出为镜像
    action_body = {
        "os-volume_upload_image": {
            "image_name": "cinder_image",
            "disk_format": "raw",
            "container_format": "bare",
            "force": True
        }
    }
    resp = conn.block_storage.post(f"/volumes/{volume.id}/action", json=action_body)
    image_id = resp.json()['os-volume_upload_image']['image_id']
    wait_for_resource_status(conn, 'image', image_id, 'active')
    print(f"镜像导出成功: {image_id}")

6. 云主机管理（Python SDK）#

使用 openstacksdk 创建云主机，代码更简洁。

import openstack, json




def create_server():
    conn = openstack.connect(
        auth_url='http://<controller>:5000/v3',
        user_domain_name='demo', username='admin', password='000000'
    )


    # 删除同名云主机
    existing = conn.compute.find_server('server001')
    if existing:
        conn.compute.delete_server(existing.id)
        print("已删除同名云主机")


    # 获取资源引用
    flavor = conn.compute.find_flavor('m1.tiny')
    image = conn.image.find_image('cirros001')
    network = conn.network.find_network('net')


    # 创建云主机
    server = conn.compute.create_server(
        name='server001',
        flavor_id=flavor.id,
        image_id=image.id,
        networks=[{'uuid': network.id}]
    )
    server = conn.compute.wait_for_server(server)
    print(f"云主机创建成功: {server.id}")
    print(json.dumps(server.to_dict(), default=str, indent=2))

7. SDK 封装管理类#

基于 openstacksdk 封装常用操作，方便在其他脚本中复用。

class ServerManager:
    def __init__(self, conn):
        self.conn = conn


    def list_servers(self):
        return list(self.conn.compute.servers())


    def get_server(self, name):
        return self.conn.compute.find_server(name)


    def create_server(self, name, image_name, flavor_name, network_name):
        flavor = self.conn.compute.find_flavor(flavor_name)
        image = self.conn.image.find_image(image_name)
        network = self.conn.network.find_network(network_name)
        server = self.conn.compute.create_server(
            name=name, flavor_id=flavor.id,
            image_id=image.id, networks=[{'uuid': network.id}]
        )
        return self.conn.compute.wait_for_server(server)


    def delete_server(self, name):
        server = self.get_server(name)
        if server:
            self.conn.compute.delete_server(server.id)
            return True
        return False




class ImageManager:
    def __init__(self, conn):
        self.conn = conn


    def list_images(self):
        return list(self.conn.image.images())


    def get_image(self, name):
        return self.conn.image.find_image(name)




class FlavorManager:
    def __init__(self, conn):
        self.conn = conn


    def list_flavors(self):
        return list(self.conn.compute.flavors())


    def get_flavor(self, name):
        return self.conn.compute.find_flavor(name)




class NetworkManager:
    def __init__(self, conn):
        self.conn = conn


    def list_networks(self):
        return list(self.conn.network.networks())


    def get_network(self, name):
        return self.conn.network.find_network(name)

8. 命令行工具开发#

用户管理 CLI（argparse + RESTful API）#

import argparse, json, sys
from api_user_manager import KeystoneUserManager, get_auth_token


def main():
    parser = argparse.ArgumentParser(description='OpenStack 用户管理工具')
    sub = parser.add_subparsers(dest='command')


    # create
    p_create = sub.add_parser('create', help='创建用户')
    p_create.add_argument('-i', '--input', required=True, help='JSON 格式用户信息')


    # getall
    p_getall = sub.add_parser('getall', help='查询所有用户')
    p_getall.add_argument('-o', '--output', help='输出文件路径')


    # get
    p_get = sub.add_parser('get', help='查询单个用户')
    p_get.add_argument('-n', '--name', required=True, help='用户名')
    p_get.add_argument('-o', '--output', help='输出文件路径')


    # delete
    p_del = sub.add_parser('delete', help='删除用户')
    p_del.add_argument('-n', '--name', required=True, help='用户名')


    args = parser.parse_args()
    headers = get_auth_token("192.168.1.10", "demo", "admin", "000000")
    mgr = KeystoneUserManager(headers, "192.168.1.10")


    if args.command == 'create':
        info = json.loads(args.input)
        print(mgr.create_user(info['name'], info['password'], info.get('description', '')))
    elif args.command == 'getall':
        result = mgr.list_users()
        if args.output:
            with open(args.output, 'w') as f:
                f.write(result)
        else:
            print(result)
    elif args.command == 'get':
        uid = mgr.get_user_id(args.name)
        if uid:
            result = mgr.get_user(uid)
            if args.output:
                with open(args.output, 'w') as f:
                    json.dump(result, f, indent=2)
            else:
                print(json.dumps(result, indent=2))
    elif args.command == 'delete':
        print(mgr.delete_user(args.name))


if __name__ == '__main__':
    main()

Flavor 管理 CLI#

Terminal window
# 创建
python flavor_manager.py create -n test-flavor -m 1024 -v 2 -d 20 -id 100


# 查询所有
python flavor_manager.py getall


# 查询单个
python flavor_manager.py get -id 100


# 删除
python flavor_manager.py delete -id 100

9. MySQL 数据库管理（SQLAlchemy）#

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker


Base = declarative_base()




class Member(Base):
    __tablename__ = 'member'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(256))
    level = Column(Integer)
    place = Column(String(256))




class DBManager:
    def __init__(self, db_name, host='<host>', password='000000'):
        self.engine = create_engine(f"mysql+pymysql://root:{password}@{host}:3306/{db_name}")
        self.Session = sessionmaker(bind=self.engine)


    def create_table(self):
        Base.metadata.create_all(self.engine)
        print("表创建成功")


    def drop_table(self):
        Base.metadata.drop_all(self.engine)
        print("表删除成功")


    def add_member(self, name, level, place):
        session = self.Session()
        member = Member(name=name, level=level, place=place)
        session.add(member)
        session.commit()
        session.close()


    def get_all_members(self):
        session = self.Session()
        members = session.query(Member).all()
        session.close()
        return members
OpenStack平台Python运维开发
https://blog.xeu.asia/posts/openstack-python-ops/
作者
Xeu
发布于
2025-03-20
许可协议
CC BY-NC-SA 4.0
X-wrt 锐捷认证
K8S容器资源Python运维开发
© 2026 Xeu. All Rights Reserved. Astro & Fuwari