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

• Reinderien: That's a lot of parameters. How is this used realisticically? Will those parameters change every time, or could they be put into a config file?
• Reinderien: Also, what warnings appear that you decided to disable?
• Reinderien: Why are you still using verify=False?
• Reinderien: What does the _c suffix stand for?
• Reinderien: Great, but that doesn't answer my question. Does the "particular customer circuit" change all the time, or are there collections of known customer parameters that are stable enough to be stored in a file?

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)

• pythontestuser: Excellent response as expected Reinderien. Thanks for your time and knowledge. I understand all elements that you have stated but for some I will need to go away investigate further to get a more thorough understanding.
• pythontestuser: Hi @Reinderien, I went through the theory of you're suggestions before I applied and have a good 90% understanding now. Thank you again, I've learned more going over this code than a lot of random blogs. The code you presented works successfully, however after the 200 OK success message, I get a TypeError: Object of type Response is not JSON serializable. If I make a change to line 110 from print(json.dumps(response, indent=4)) to print(json.dumps(response.json(), indent=4)) I then don't get the Type Error and get the display as expected. Is my action correct or have I just fudged it?
• pythontestuser: Apologies I couldn't show full response outputs due to character limits
• Reinderien: @pythontestuser My mistake; edited.
• pythontestuser: no issues what so ever. I just wanted to make sure I'd not fudged it. Thanks again for your time and efforts. All the best and potentially catch you soon.