Python script to POST requests to a networking device

Question

Request to review my code please and see if I am following the best practices.

This is sending a POST operation via a Python script using the requests library to a networking cloud orchestration endpoint

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())

Credit to @Reinderien who helped me formulate best practices for GET requests which I have adopted into my POST script (with a few minor tweaks) above.

Successful response shown below

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'}}

Answer 1

Disable hard certificate failures, fine (temporarily). But do not disable_warnings. They're there to remind you that you're doing a bad thing. You're a network engineer, so you appreciate more than I do the hazards of handicapping TLS.

Remove your _c suffixes. If I understood correctly, these are used as a rudimentary form of source control to show that this is the third incarnation of the script, but this is really not a good idea.

Given the context of this question, which is that the script is intended for eventual automated reuse, I'm going to recommend that you convert your inputs to argparse parameters. If you're lucky, this script might eventually be drop-in reused with no modification needed.

Your payload is prematurely serialised. I realise that the documentation told you to do this, but the documentation is wrong. You shouldn't be manually writing out a JSON representation in a string constant and then passing that to data=; start with a dictionary and pass that to json=. In other words, tell Requests to do the hard work for you.

You have a main method (good), but it's failed to capture half of your imperative instructions, particularly your inputs. Constants can remain in the global namespace, but thing-doers should be in functions.

You have a sys.exit(main()) which is a typical pattern to expose a numeric return code to the shell, but you've only half-implemented it. main needs to return an integer for this to work. You should return 0 on success, and different non-zero values based on various failures you see. One of these failures should be a non-200-series HTTP response code.

Beside your post(), you should include a link to the API documentation.

For automation compatibility, the only output going to stdout should be JSON. The rest - HTTP statuses and error codes - should go to stderr.

This use case has already exceeded the forgiving circumstances in the previous question that allowed for simplified exception handling. You should only catch the exceptions you anticipate. Everything else should be left exceptional.

As a refactor, consider writing:

• more subroutines
• an Enum to self-document your process status codes
• argparse support

Suggested

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)