Скрипт Python для POST-запросов к сетевому устройству

Question 1

Запросите просмотр моего кода, пожалуйста, и посмотрите, следую ли я лучшим практикам.

Это отправка операции POST через скрипт Python с использованием библиотеки запросов в конечную точку сетевой облачной оркестровки.

import requests
import sys

requests.packages.urllib3.disable_warnings()

token = input("Please enter your token: ")

dev_ip = "1.1.1.1"

service_id_c = input("Please enter the service ID: ")
out_vlan_c = int(input("Please enter the uplink outer vlan ID: "))
in_vlan_c = int(input("Please enter the uplink inner vlan ID: "))
content_provider_c = input("Please enter the Content Provider name: ")
interface_name_c = input("Please enter the Downlink Interface name: ")
remote_id_c = input("Please enter the Remote ID: ")
agent_id_c = input("Please enter the Agent Circuit ID: ")
profile_name_c = input("Please enter the Profile name: ")



payload = f"{{\r\n    \"input\": {{\r\n        \"service-context\": {{\r\n            \"service-id\": \"{service_id_c}\",\r\n            \"uplink-endpoint\": {{\r\n                \"interface-endpoint\": {{\r\n                    \"outer-tag-vlan-id\": {out_vlan_c},\r\n                    \"inner-tag-vlan-id\": {in_vlan_c},\r\n                    \"content-provider-name\": \"{content_provider_c}\"\r\n                }}\r\n            }},\r\n            \"downlink-endpoint\": {{\r\n                \"interface-endpoint\": {{\r\n                    \"outer-tag-vlan-id\": \"untagged\",\r\n                    \"inner-tag-vlan-id\": \"none\",\r\n                    \"interface-name\": \"{interface_name_c}\"\r\n                }}\r\n            }},\r\n            \"remote-id\": \"{remote_id_c}\",\r\n            \"agent-circuit-id\": \"{agent_id_c}\",\r\n            \"profile-name\": \"{profile_name_c}\"\r\n        }}\r\n    }}\r\n}}"

headers = {
    'Authorization': "Bearer " + token,
}

def main():
    try:
        with requests.post(
            url=f"https://{dev_ip}/api/restconf/operations/cloud-platform-orchestration:create",
            headers=headers,
            data=payload,
            verify=False,
            timeout=10,
        ) as response:
            represt_c = response.json()
    except Exception as e:
        print(f'An error occurred, please investigate further: {e!r}')
    else:
        print(response.status_code, response.reason, response.url)
        print(represt_c)

if __name__ == '__main__':
    sys.exit(main())

Спасибо @Reinderien, который помог мне сформулировать лучшие практики для запросов GET, которые я внедрил в свой сценарий POST (с несколькими незначительными изменениями) выше.

Успешный ответ показан ниже

Please enter your token: ******************************
Please enter the service ID: ABCD-ABC2525
Please enter the uplink outer vlan ID: 3060
Please enter the uplink inner vlan ID: 1060
Please enter the Content Provider name: COMPANY-SERVICE-B50
Please enter the Downlink Interface name: ABC123456_ETH_20
Please enter the Remote ID: test897
Please enter the Agent Circuit ID: test897
Please enter the Profile name: Data Service
200 OK https://192.168.1.1/api/restconf/operations/abc-cloud-platform-orchestration:create
{'output': {'completion-status': 'in-progress',
            'service-id': 'ADTN-ADTN2525',
            'status': 'creating',
            'timestamp': '2022-05-16T2:41:11.371020',
            'trans-id': 'ed2667f3-629384-22734-t334-07d345551b93'}}

Question 2

Отключите жесткие сбои сертификатов, хорошо (временно). Но не надо disable_warnings. Они здесь, чтобы напомнить вам, что вы делаете плохие вещи. Вы сетевой инженер, поэтому вы больше, чем я, понимаете опасность ограничения TLS.

Удалите свой _c суффиксы. Если я правильно понял, они используются как рудиментарная форма контроля версий, чтобы показать, что это третья инкарнация скрипта, но на самом деле это не очень хорошая идея.

Учитывая контекст этого вопроса, который заключается в том, что сценарий предназначен для возможного автоматического повторного использования, я собираюсь порекомендовать вам преобразовать ваш inputс к argparse параметры. Если вам повезет, этот скрипт в конечном итоге может быть повторно использован без каких-либо изменений.

Ваш payload преждевременно сериализуется. Я понимаю, что документация говорит вам сделать это, но документация неверна. Вы не должны вручную записывать представление JSON в строковой константе, а затем передавать это в data=; начните со словаря и передайте его json=. Другими словами, попросите Requests сделать всю тяжелую работу за вас.

У тебя есть main метод (хороший), но он не смог уловить половину ваших императивных инструкций, особенно inputс. Константы могут оставаться в глобальном пространстве имен, но исполнители должны находиться в функциях.

У тебя есть sys.exit(main()) это типичный шаблон для представления числового кода возврата в оболочку, но вы реализовали его только наполовину. main должен вернуть целое число, чтобы это работало. Вы должны вернуть 0 в случае успеха и разные ненулевые значения в зависимости от различных сбоев, которые вы видите. Один из этих сбоев должен быть кодом ответа HTTP, отличным от серии 200.

Рядом с вашим post()вы должны включить ссылку на документацию по API.

Для совместимости с автоматикой единственным выходом будет stdout должен быть JSON. Остальное — HTTP-статусы и коды ошибок — следует отправлять в stderr.

Этот вариант использования уже превысил прощающие обстоятельства в предыдущем вопросе, что позволило упростить обработку исключений. Вы должны ловить только те исключения, которые ожидаете. Все остальное следует оставить исключительным.

В качестве рефакторинга рассмотрите возможность написания:

больше подпрограмм
ан Enum для самостоятельного документирования кодов состояния вашего процесса
argparse поддерживать

Предложенный

from argparse import ArgumentParser, Namespace
from enum import Enum
from json import JSONDecodeError
from typing import Any

import json
import requests
import sys


class ProcessStatus(Enum):
    OK = 0
    PYTHON_DEFAULT_ERROR = 1
    IO_ERROR = 2
    JSON_ERROR = 3
    HTTP_ERROR = 4


def get_args() -> Namespace:
    parser = ArgumentParser(
        description='Fill out circuit parameters and send an orchestration create command to the cloud.',
    )
    parser.add_argument('--host', '-s', default="1.1.1.1",
                        help='Orchestration server host')
    parser.add_argument('--token', '-t', required=True,
                        help='Bearer token for authorisation to the cloud service')
    parser.add_argument('--service-id', '-v', required=True,
                        help='Service ID of the circuit')
    parser.add_argument('--out-vlan', '-o', required=True, type=int,
                        help='ID of the outgoing VLAN')
    parser.add_argument('--in-vlan', '-i', required=True, type=int,
                        help='ID of the incoming VLAN')
    parser.add_argument('--content-provider', '-c', required=True,
                        help='Content provider name')
    parser.add_argument('--downlink-interface', '-d', required=True,
                        help='Downlink interface name')
    parser.add_argument('--remote-id', '-r', required=True,
                        help='Remote ID')
    parser.add_argument('--agent-circuit', '-a', required=True,
                        help='Agent circuit ID')
    parser.add_argument('--profile', '-p', required=True,
                        help='Profile name')

    return parser.parse_args()


def fill_payload(args: Namespace) -> dict[str, Any]:
    payload = {
        'input': {
            'service-context': {
                'service-id': args.service_id,
                'uplink-endpoint': {
                    'interface-endpoint': {
                        'outer-tag-vlan-id': args.out_vlan,
                        'inner-tag-vlan-id': args.in_vlan,
                        'content-provider-name': args.content_provider,
                    }
                },
                'downlink-endpoint': {
                    'interface-endpoint': {
                        'outer-tag-vlan-id': 'untagged',
                        'inner-tag-vlan-id': 'none',
                        'interface-name': args.downlink_interface,
                    }
                },
                'remote-id': args.remote_id,
                'agent-circuit-id': args.agent_circuit,
                'profile-name': args.profile,
            }
        }
    }
    return payload


def post(host: str, token: str, payload: dict[str, Any]) -> requests.Response:
    headers = {
        'Authorization': 'Bearer ' + token,
        'Content-Type': 'application/yang-data+json',
        'Accept': 'application/yang-data+json',
    }

    # See "Create Bundle":
    # https://documenter.getpostman.com/view/2389999/TVsrGVaa#071c1413-852b-467d-9049-8f19fb757d9b

    return requests.post(
        url=f'https://{host}/api/restconf/operations/cloud-platform-orchestration:create',
        headers=headers,
        json=payload,
        verify=False,
        timeout=10,
    )


def main() -> ProcessStatus:
    args = get_args()
    payload = fill_payload(args)

    try:
        response = post(host=args.host, token=args.token, payload=payload)
    except IOError as e:
        print(f'An I/O error occurred: {e!r}', file=sys.stderr)
        return ProcessStatus.IO_ERROR

    print(response.status_code, response.reason, response.url, file=sys.stderr)

    try:
        print(json.dumps(response.json(), indent=4))
    except JSONDecodeError:
        print('The response could not be decoded:\n', response.text, file=sys.stderr)
        return ProcessStatus.JSON_ERROR

    if response.ok:
        return ProcessStatus.OK
    return ProcessStatus.HTTP_ERROR


if __name__ == '__main__':
    sys.exit(main().value)

Скрипт Python для POST-запросов к сетевому устройству

1 ответ
1

Предложенный

Добавить комментарий Отменить ответ

1 ответ 1

Предложенный

Похожие записи:

Добавить комментарий Отменить ответ

1 ответ
1