2
\$\begingroup\$

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.

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

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

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.