# Modify original JSON few times and post http request

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

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:
count += 1
r = requests.get(url.format(ip))
data = r.json()
if (
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}")

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
# 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
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


In the case of verify action failure:

python test.py dev master-instace test-config-123.tgz
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.

• This is a follow-up question – Grajdeanu Alex Dec 10 '20 at 9:10
• Just edited it in my question as well. – AndyP Dec 10 '20 at 14:23

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


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],
):
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 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'.
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.
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.',
)
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()

• 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? – AndyP Dec 12 '20 at 18:46
• @AndyP Can you provide meaningful argv parameters for me to test this, to be able to provide an example? – Reinderien Dec 12 '20 at 19:10
• 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. – AndyP Dec 12 '20 at 19:18
• @AndyP Is it actually master-instace, or is it master-instance ? – Reinderien Dec 12 '20 at 19:20
• You should think about fixing the spelling. Anyway, I'm writing up an example. – Reinderien Dec 12 '20 at 19:30