Modify original JSON few times and post http request

Question

Follow up question from here.

I am working on a project where I need to do below things in Python:

• Take few input parameters like - environmentName, instanceName, configName from command line.

• Get JSON string as the response by calling service url. For example - Below is the JSON I will get:

  {"flag":false, "configName":"abc-123.tar"}

• Modify the above JSON by passing configName passed from the command line. So if configName is passed as abc-124.tar then my modified JSON will be:

  {"flag":false, "configName":"abc-124.tar"}

• Now there are two actions - download and verify. For download action - I will modify above JSON like below and then post this new modified JSON to same service.

  {"flag":false, "configName":"abc-124.tar", "action":"download"}

• Once all machines have downloaded successfully then I will modify my JSON again for verify action like below and then post this new modified JSON again to same service.

  {"flag":false, "configName":"abc-124.tar", "action":"verify"}

• Now I will verify whether all files looks good or not. But if download action failed then I will not do verify action.

I am calling another endpoint to validate whether each action (download or verify) completed successfully or not. This logic is done in validate method.

Below is my code which does that and it works fine:

import json
import sys
import requests
ACTIONS = ['download', 'verify']
ENDPOINT = "http://consul.{}.demo.com/"
CONFIG_PATH = "v1/kv/app-name/{}/config"
STATUS_PATH = "v1/kv/app-name/{}/status/{}"
CATALOG_PATH = "v1/catalog/service/app-name?ns=default&tag={}"
RAW = "?raw"
def update(url, json_):
 try:
 r = requests.put(url, data=json_, headers={'Content-type': 'application/json'})
 r.raise_for_status()
 return True
 except requests.exceptions.HTTPError as http_error:
 print(f"Http Error: {http_error}")
 except requests.exceptions.ConnectionError as connection_error:
 print(f"Error Connecting: {connection_error}")
 except requests.exceptions.Timeout as timeout_error:
 print(f"Timeout Error: {timeout_error}")
 except requests.exceptions.RequestException as request_error:
 print(f"Ops: Something Else: {request_error}")
 return False
def validate(environment, instance, config, action):
 flag = True
 catalog_url = ENDPOINT.format(environment) + CATALOG_PATH.format(instance)
 response = requests.get(catalog_url)
 url = ENDPOINT.format(environment) + STATUS_PATH.format(instance) + RAW
 json_array = response.json()
 count = 0
 for x in json_array:
 ip = x['Address']
 count += 1
 r = requests.get(url.format(ip))
 data = r.json()
 # validate download action here
 if (
 action == 'download' 
 and "isDownloaded" in data 
 and data['isDownloaded'] 
 and "currentCfg" in data 
 and data['currentCfg'] == config
 ):
 continue
 # validate verify action here
 elif (
 action == 'verify' and "activeCfg" in data 
 and data['activeCfg'] == config
 ):
 continue
 else:
 flag = False
 print(f"Failed to {action} on {ip}")
 if flag and action == 'download':
 print(f"Downloaded successfully on all {count} machines")
 elif flag and action == 'verify':
 print(f"Verified successfully on all {count} machines")
 return flag
def main():
 environment = sys.argv[1]
 instance = sys.argv[2]
 new_config = sys.argv[3]
 # make url for get api to get JSON config
 url = ENDPOINT.format(environment) + CONFIG_PATH.format(instance)
 response = requests.get(url + RAW)
 # get JSON string from my service
 data = json.loads(response.content)
 # modify json with config being passed from command line
 data['remoteConfig'] = new_config
 
 # now for each action, update json and then validate whether that 
 # action completed successfully or not.
 for action in ACTIONS:
 data['action'] = action
 if update(url, json.dumps(data)):
 if validate(environment, instance, new_config, action):
 continue
 else:
 print(f"{action} action failed")
 return
 else:
 print("failed to update")
 return
if __name__ == "__main__":
 try:
 sys.exit(main())
 except KeyboardInterrupt:
 print("\nCaught ctrl+c, exiting")

Problem Statement

I have my above code which works fine so opting for code review to see if there is any better way to do the above things efficiently and cleanly. I am new to Python so I might have made lot of mistakes in designing it. Given this code will be running in production wanted to see what can be done to improve it?

The idea is something like this when I run it from the command line.

If everything is successful then it should look like this:

python test.py dev master-instace test-config-123.tgz
downloaded successfully on all 10 machines
verified successfully on all 10 machines
config pushed successfully!!

In case of download action failure:

python test.py dev master-instace test-config-123.tgz
failed to download on 1.2.3.4
failed to download on 4.2.3.4
download action failed

In the case of verify action failure:

python test.py dev master-instace test-config-123.tgz
downloaded successfully on all 10 machines
failed to verify on 1.2.3.4
failed to verify on 4.2.3.4
verify action failed

or if there is any better way to make the output cleaner - I am open to the suggestion as well.

• Grajdeanu Alex: This is a follow-up question
• AndyP: Just edited it in my question as well.

Answer 1

Manual serialization and content type

Try to avoid this:

requests.put(url, data=json_, headers={'Content-type': 'application/json'})
 

Read https://requests.readthedocs.io/en/master/user/quickstart/#more-complicated-post-requests . It serializes the JSON for you and sets the content type for you if you use the json= kwarg instead, passing a dictionary or list.

Variable naming

Rename flag to something like is_valid.

Format binding

There's a neat trick that is sometimes used in Python for constructs like this:

ENDPOINT = "http://consul.{}.demo.com/"
CONFIG_PATH = "v1/kv/app-name/{}/config"
STATUS_PATH = "v1/kv/app-name/{}/status/{}"
CATALOG_PATH = "v1/catalog/service/app-name?ns=default&tag={}"
# ...
url = ENDPOINT.format(environment) + CONFIG_PATH.format(instance)

Instead consider

endpoint = "http://consul.{}.demo.com/".format
config_path = "v1/kv/app-name/{}/config".format
status_path = "v1/kv/app-name/{}/status/{}".format
catalog_path = "v1/catalog/service/app-name?ns=default&tag={}".format
# ...
url = endpoint(environment) + config_path(instance)

These variables become function references, so that you can only call format on the referenced string and nothing else.

if/else

This thing in a loop:

 if update(url, json.dumps(data)):
 if validate(environment, instance, new_config, action):
 continue
 else:
 print(f"{action} action failed")
 return
 else:
 print("failed to update")
 return

seems like it would be more easily expressed as

if not (
 update(url, json.dumps(data)) and
 validate(environment, instance, new_config, action)
):
 print(f"{action} action failed")
 return

However, a better way to make use of exceptions is:

• Stop returning booleans from your inner functions
• Move your try to outside of the for loop in main
• Also move your except and print statements outside of the for loop in main

Then you will not need a continue - simply loop, and outside of the loop if anything is caught, the loop will have been broken for you.

Return codes

You do

sys.exit(main())

but you never return an integer from main. Either drop the exit, or return meaningful integer codes from main.

• Remove your if applied to update and validate

Proposed

Since your API is hosted internally I wasn't able to test this very far.

from argparse import ArgumentParser
from typing import Tuple, List, Dict
from requests import Session
class NetworkError(Exception):
 pass
class ValidationError(Exception):
 pass
ENDPOINT_PATH = 'http://consul.{env:}.demo.com/v1/'
make_config_url = (
 ENDPOINT_PATH + 'kv/app-name/{instance:}/config?raw'
).format
make_status_url = (
 ENDPOINT_PATH + 'kv/app-name/{instance:}/status/{ip:}?raw'
).format
make_catalog_url = (
 ENDPOINT_PATH + 'catalog/service/app-name'
).format
def validate(
 session: Session, env: str, instance: str, new_config: str, action: str,
 machine: Dict[str, object],
):
 ip = machine['Address']
 with session.get(
 make_status_url(env=env, instance=instance, ip=ip)
 ) as response:
 response.raise_for_status()
 status = response.json()
 prefix = f'{action} failed for {ip}: '
 if action == 'download':
 if not status.get('isDownloaded'):
 raise ValidationError(prefix + 'not downloaded')
 if status.get('currentCfg') != new_config:
 raise ValidationError(prefix + 'configuration mismatch')
 elif action == 'verify' and status.get('activeCfg') != new_config:
 raise ValidationError(prefix + 'configuration mismatch')
def validate_all(
 session: Session, env: str, instance: str, new_config: str, action: str
) -> Tuple[
 int, # Succeeded machines
 int, # Total machines
 List[str], # Error messages
]:
 """
 This will get list of all ipAddresses for that 'instance' in that 'environment'.
 And then check whether each machine/ipAddress from that list have downloaded and
 verified successfully. Basically for each 'action' type I need to validate to make
 sure each of those 'actions' completed successfully. If successful at the end then print out the message.
 Returns the number of machines.
 """
 with session.get(
 make_catalog_url(env=env),
 params={'ns': 'default', 'tag': instance}
 ) as response:
 response.raise_for_status()
 machines = response.json()
 n_succeeded = 0
 n_total = len(machines)
 messages = []
 for machine in machines:
 try:
 validate(session, env, instance, new_config, action, machine)
 except Exception as e:
 messages.append(str(e))
 n_succeeded += 1
 return n_succeeded, n_total, messages
def reconfigure(env: str, instance: str, new_config: str):
 config_url = make_config_url(env=env, instance=instance)
 with Session() as session:
 try:
 with session.get(config_url) as response:
 response.raise_for_status()
 data = response.json()
 except Exception as e:
 raise NetworkError('Failed to get original config') from e
 # modify json with config being passed from command line
 data['remoteConfig'] = new_config
 # now for each action, update json and then validate whether that
 # action completed successfully or not.
 for action in ('download', 'verify'):
 data['action'] = action
 try:
 with session.put(config_url, json=data) as response:
 response.raise_for_status()
 except Exception as e:
 raise NetworkError(f'Failed to update for action {action}') from e
 n_succeeded, n_total, errors = validate_all(session, env, instance, new_config, action)
 
 print(f'Action {action} validated for {n_succeeded}/{n_total} machines')
 print('\n'.join(errors))
def main():
 parser = ArgumentParser(
 description='Reconfigure a Consul service to do some magic.',
 )
 parser.add_argument('-e', '--env')
 parser.add_argument('-i', '--instance')
 parser.add_argument('-c', '--config')
 args = parser.parse_args()
 try:
 reconfigure(args.env, args.instance, args.config)
 except KeyboardInterrupt:
 pass
 except Exception as e:
 print(e, end='')
 if e.__cause__ is None:
 print()
 else:
 print(':', e.__cause__)
 exit(1)
if __name__ == '__main__':
 main()

• AndyP: Thanks for great suggestion. Learned something new. I will read about the link you shared about the post requests. But I am kinda confuse on this suggestion However, a better way to make use of exceptions is:. Do you think you can provide can example basis on my code so that I can understand properly on how to do it cleanly? Also do you see any design issues in my above code? Like it can be rewritten in better ways if possible?
• Reinderien: @AndyP Can you provide meaningful argv parameters for me to test this, to be able to provide an example?
• AndyP: In my case, argv parameters are gonna be exactly like what I have in my question python test.py dev master-instace test-config-124.tgz. Testing for you might be hard as it involves working with Consul service. As of now I call consul service to get JSON string and then change configName in the original JSON string and update it on Consul for each action type. If download action works fine successfully on each machine then I do same thing for verify action. Let me know if anything is unclear and then I can try to help you out.
• Reinderien: @AndyP Is it actually master-instace, or is it master-instance ?
• Reinderien: You should think about fixing the spelling. Anyway, I'm writing up an example.