Что это?

Небольшая асинхронная библиотека на основе aiohttp, содержащая несколько публичных классов для работы с API VMmanager 6.

• AuthSession
• DnsProxySession
• IpSession
• VmSession

Каждый класс наследуется от базового класса BaseSession, который обарачивает aiohttp.ClientSession и действует как менеджер контекста.

В библиотеке уже реализовано формирование удобных конечных точек. Библиотека декодирует ответы JSON, а также вызывает исключение ClientResponseError, если статус ответа 400 или выше.

Также есть несколько готовых методов основных действий:

AuthSession

• get_token
• get_key
• whoami

VmSession

• get_task
• get_task_by_consul_id
• host_create
• host_delete
• host_edit

Как это использовать?

Каждое доступное направление API VMmanager 6 имеет свой класс. При инициализации класса вы можете указать базовый URL-адрес и проверку SSL-сертификата. Все остальные параметры переопределяют поведение aiohttp.ClientSession. Подробнее можно прочитать в официальной документации. aiohttp

Каждый класс имеет методы, упрощающие создание запросов API. Их имена такие же, как у методов aiohttp.ClientSession. Реализованные в настоящее время методы:

• get
• post
• delete

Чтобы начать генерировать запросы, импортируйте необходимый класс и используйте синтаксис контекстного менеджера

Установка

Требования

• python = ^3.9

В ручную

• установите зависимости: pip install aiohttp
• скопируйте aiovmmanager package в ваш проект

Используя pip

• pip install iovmmanager

Примеры

• получить токен авторизации этот токен нужно будет добавлять в заголовок каждого запроса в будущем

import asyncio
from aiovmmanager import AuthSession


async def main():
    url = 'https://vm6.example.com'
    headers = {}

    # pass to AuthSession ssl=False if you have a self-signed or invalid certificate
    async with AuthSession(url=url) as session:
        response = await session.get_token(
            email='admin@example.com',
            password='password',
        )
        # set x-xsrf-token header:
        headers['x-xsrf-token'] = response.get('token', '')
        print(headers)


if __name__ == "__main__":
    asyncio.run(main())

• сквозная авторизация по ключу

import asyncio
from aiovmmanager import AuthSession


async def main():
    url = 'https://vm6.example.com'
    headers = {'x-xsrf-token': 'the token can be obtained in the example above'}

    # pass to AuthSession ssl=False if you have a self-signed or invalid certificate
    async with AuthSession(url=url, headers=headers) as session:
        response = await session.get_key(email_or_id='admin@example.com')
        print(response)


if __name__ == "__main__":
    asyncio.run(main())

• Создать 3 виртуальные машины асинхронно

import asyncio
from aiovmmanager import VmSession


vm_template = {
    "name": "aiovmmgr",
    "os": 8211,
    "password": "root-password",
    "send_email_mode": "default",
    "cluster": 27,
    "preset": 1,
    "disks": [
        {
            "boot_order": 1,
            "size_mib": 10240,
            "tags": [],
            "storage": 61
        }
    ],
    "comment": "test vm from aiovmmgr",
    "account": 124,
    "node": 64,
    "custom_interfaces": [
        {
            "model": "virtio",
            "is_main_network": True,
            "bridge_id": 97,
            "ip_count": 1,
            "ippool": 111
        }
    ],
    "domain": "aiovmmgr.example.com"
}


async def main():
    url = 'https://vm6.example.com'
    headers = {'x-xsrf-token': 'the token can be obtained in the first example'}

    async with VmSession(url=url, headers=headers) as session:
        # create a list of three coroutines
        # of course in a real example host_params contains three different vms
        tasks = [session.host_create(host_params=vm_template) for _ in range(3)]
        # send a request to create virtual machines asynchronously
        results = await asyncio.gather(*tasks)

        for result in results:
            print(result)


if __name__ == "__main__":
    asyncio.run(main())

• получить задание по consul_id

Все задачи, которые долгое время выполняет платформа, попадают в consul. В ответ приходит идентификатор такой задачи. В дальнейшем задача перейдет в диспетчер задач и будет обработана и завершена. Номер задачи можно узнать в диспетчере задач, запросив информацию по consul_id. В методе настроен фильтр точного соответствия по consul_id.

import asyncio
from aiovmmanager import VmSession


async def main():
    url = 'https://vm6.example.com'
    headers = {'x-xsrf-token': 'the token can be obtained in the first example'}

    consul_id = 1488228  # example consul_id

    async with VmSession(url=url, headers=headers) as session:
        response = await session.get_task_by_consul_id(consul_id=consul_id)
        print(response)

if __name__ == "__main__":
    asyncio.run(main())

• любой запрос в соответствии с документацией API VMmanager 6

Поскольку все классы наследуются от BaseSession, который является обертку aiohttp.ClientSession, вы можете делать запросы к любым обработчикам.

import asyncio
from aiovmmanager import AuthSession, VmSession


async def main():
    url = 'https://vm6.example.com'
    headers = {'x-xsrf-token': 'the token can be obtained in the first example'}

    # -- Snip
    async with AuthSession(url=url, headers=headers) as session:
        # show active admins:
        list_admins = await session.get(
            url='/user',
            params={"where": "((roles+CP+'%@admin%')+AND+(state+EQ+'active'))"}
        )
        print(list_admins)
    # -- Snip

    # -- Snip
    async with VmSession(url=url, headers=headers) as session:
        # force restart vm by id:
        host_id = 228
        restart_vm = await session.post(
            url=f'/host/{host_id}/restart',
            json={'force': True}
        )
        print(restart_vm)
    # -- Snip

if __name__ == "__main__":
    asyncio.run(main())