Skip to main content
Code Review

Return to Answer

added 8 characters in body
Source Link
Reinderien
Reinderien
  • 71.2k
  • 5
  • 76
  • 257
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)

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

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)
added 206 characters in body
Source Link
Reinderien
Reinderien
  • 71.2k
  • 5
  • 76
  • 257

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.

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.

Source Link
Reinderien
Reinderien
  • 71.2k
  • 5
  • 76
  • 257

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.

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