0
\$\begingroup\$

I have written a rudimentary code in Python:

  1. That reads the API collection from Swagger/Postman and extracts the payload, link, and method
  2. It creates the response with the token(Generated from the token link)
  3. It records the response in a CSV.

Need help in further formatting the code in terms of

  1. creating functions.

  2. Reducing lines of code.

  3. Performance tuning using multithreading

     from urllib.request import urlopen
     import requests
     from bs4 import BeautifulSoup
     import json
     from jsonpath_ng import jsonpath, parse
    
    
     def GenerateToken():
         link = "<link>"
         page = urlopen(link)  # Link to generate token
         contents = page.read()
         soup = BeautifulSoup(contents)
         shred = list(soup.stripped_strings)
         token = shred[2]  # gets the token
         return token
    
    
     with open("TestResults.csv", "w") as f_object:
         fields = ["API_Endpoint", "Method", "Status_Code", "Message", "Timestamp"]
         writer = csv.DictWriter(f_object, fieldnames=fields)
         writer.writeheader()
         headers = {"Authorization": "Bearer {}".format(GenerateToken())}
         with open("postman_collection.json") as f:
             data = json.load(f)
             for res in data["item"]:
    
                 for d in res["request"]:
    
                     if "GET" in res["request"]["method"]:
                         payload = {}
                         url = res["request"]["url"]["raw"]
                         method = res["request"]["method"]
    
                         try:
                             response = requests.get(url, headers=headers, data=payload)
    
                             if response.status_code == 200:
                                 rows1 = [
                                     {
                                         "API_Endpoint": url,
                                         "Method": method,
                                         "Status_Code": response.status_code,
                                         "Message": "OK",
                                         "Timestamp": datetime.datetime.utcnow(),
                                     }
                                 ]
                                 writer.writerows(rows1)
                             # writer.writerows(rows1)
                             else:
                                 rows2 = [
                                     {
                                         "API_Endpoint": url,
                                         "Method": method,
                                         "Status_Code": response.status_code,
                                         "Message": response.text,
                                         "Timestamp": datetime.datetime.utcnow(),
                                     }
                                 ]
                                 writer.writerows(rows2)
                         # print(response.json().get("message"))
                         except requests.ConnectionError:
                             print("failed to connect")
    
                     elif "POST" in res["request"]["method"]:
                         url = res["request"]["url"]["raw"]
                         method = res["request"]["method"]
                         payload = res["request"]["body"]["raw"]
                         try:
                             # response = requests.request("GET",url,headers=headers, data=payload)
                             response = requests.request(
                                 "POST", url, headers=headers, data=payload
                             )
                             # print(response.json())
                             if response.status_code == 200:
                                 rows3 = [
                                     {
                                         "API_Endpoint": url,
                                         "Method": method,
                                         "Status_Code": response.status_code,
                                         "Message": "OK",
                                         "Timestamp": datetime.datetime.utcnow(),
                                     }
                                 ]
                                 writer.writerows(rows3)
                             else:
                                 rows4 = [
                                     {
                                         "API_Endpoint": url,
                                         "Method": method,
                                         "Status_Code": response.status_code,
                                         "Message": response.text,
                                         "Timestamp": datetime.datetime.utcnow(),
                                     }
                                 ]
                                 writer.writerows(rows4)
                                 # print(response.json().get("message"))
                         except requests.ConnectionError:
                             print("failed to connect")
    
                     else:
                         url = res["request"]["url"]["raw"]
                         method = res["request"]["method"]
                         payload = res["request"]["body"]["raw"]
                         try:
                             # response = requests.request("GET",url,headers=headers, data=payload)
                             response = requests.request(
                                 "POST", url, headers=headers, data=payload
                             )
                             # print(response.json())
                             if response.status_code == 200:
                                 rows5 = [
                                     {
                                         "API_Endpoint": url,
                                         "Method": method,
                                         "Status_Code": response.status_code,
                                         "Message": "OK",
                                         "Timestamp": datetime.datetime.utcnow(),
                                     }
                                 ]
                                 writer.writerows(rows5)
                             else:
                                 rows6 = [
                                     {
                                         "API_Endpoint": url,
                                         "Method": method,
                                         "Status_Code": response.status_code,
                                         "Message": response.text,
                                         "Timestamp": datetime.datetime.utcnow(),
                                     }
                                 ]
                                 writer.writerows(rows6)
                         # print(response.json().get("message"))
                         except requests.ConnectionError:
                             print("failed to connect")
     f_object.close()
    
\$\endgroup\$
3
  • 1
    \$\begingroup\$ The title of the question should be what the code does rather than your concerns about the code. Please read How do I ask a good question. \$\endgroup\$
    – pacmaninbw
    Jan 29, 2022 at 12:38
  • 1
    \$\begingroup\$ Please do not edit the question, especially the code, after an answer has been posted. Changing the question may cause answer invalidation. Everyone needs to be able to see what the reviewer was referring to. What to do after the question has been answered. \$\endgroup\$
    – pacmaninbw
    Jan 30, 2022 at 14:05
  • 1
    \$\begingroup\$ @Durga. If you need a review on updated code, please post that as a second question, rather than editing the code in this one. \$\endgroup\$
    – user673679
    Jan 30, 2022 at 15:20

1 Answer 1

2
\$\begingroup\$

There is indeed a lot of repetition in this code.

You have 3 cases: GET, POST, other. So we have:

 if "GET" in res["request"]["method"]:
     payload = {}
     url = res["request"]["url"]["raw"]
     method = res["request"]["method"]
 elif "POST" in res["request"]["method"]:
     url = res["request"]["url"]["raw"]
     method = res["request"]["method"]
     payload = res["request"]["body"]["raw"]
 else:
     url = res["request"]["url"]["raw"]
     method = res["request"]["method"]
     payload = res["request"]["body"]["raw"]

Now that we have an simple overview we see that the whole block can be reduced as follows:

url = res["request"]["url"]["raw"]
method = res["request"]["method"]
if "GET" in res["request"]["method"]:
    payload = {}
else:
    payload = res["request"]["body"]["raw"]

There is only a slight difference between methods: the payload.

Then we proceed to the request proper:

# GET
response = requests.get(url, headers=headers, data=payload)

# POST
response = requests.request(
 "POST", url, headers=headers, data=payload
)

# other
response = requests.request(
 "POST", url, headers=headers, data=payload
)

Again, all very similar. We can build an all-purpose statement like this:

method = res["request"]["method"]
response = requests.request(method, url, headers=headers, data=payload)

There is no need for 3 different statements since they all pass the same arguments. The only thing that is different is the method, and you can pass that as an argument too.

So now that you have a single statement to issue the request, we need to look at the result. It's a simple condition of 200 or other. Thus:

if response.status_code == 200:
    result = "OK"
else:
    result = response.text

rows = [
    {
    "API_Endpoint": url,
    "Method": method,
    "Status_Code": response.status_code,
    "Message": result,
    "Timestamp": datetime.datetime.utcnow(),
    }
]
writer.writerows(rows)

As far as I can see you are repeating the same stuff 6 times. The only thing that varies is the method + the HTTP status code and the resulting response message. That can be easily simplified.

The final f_object.close() is not required since you are using the context manager (with).

Accordingly, your code can already be shortened like this (untested but the concision is hopefully becoming perceptible):

with open("TestResults.csv", "w") as f_object:
    fields = ["API_Endpoint", "Method", "Status_Code", "Message", "Timestamp"]
    writer = csv.DictWriter(f_object, fieldnames=fields)
    writer.writeheader()
    headers = {"Authorization": "Bearer {}".format(GenerateToken())}
    with open("postman_collection.json") as f:
        data = json.load(f)
        for res in data["item"]:

            for d in res["request"]:

                # common for all methods
                url = res["request"]["url"]["raw"]
                method = res["request"]["method"]

                # empty payload for GET
                if "GET" in res["request"]["method"]:
                    payload = {}
                else:
                    payload = res["request"]["body"]["raw"]

                try:
                    response = requests.request(method, url, headers=headers, data=payload)

                    if response.status_code == 200:
                        result = "OK"
                    else:
                        result = response.text

                    rows = [
                        {
                        "API_Endpoint": url,
                        "Method": method,
                        "Status_Code": response.status_code,
                        "Message": result,
                        "Timestamp": datetime.datetime.utcnow(),
                        }
                    ]
                    writer.writerows(rows)
                    # print(response.json().get("message"))
                except requests.ConnectionError:
                    print("failed to connect")

So the code is now becoming more manageable after removing duplicate code. Basically we went from 118 lines to 43.

That block of code could fit into a function, which name should reflect the actual purpose (that we don't know). Then it would be good to pass file names as arguments, rather than have them hardcoded inside the function.

If you are going to perform repeated requests against a host then you should use requests.session to improve performance and simplifying handling of state, cookies, headers etc.

To address some of your concerns: if the intention is to perform parallel processing or multithreading, there is more than one way to do that in Python. Which is best depends on your circumstances. May I suggest one link as a tutorial: Python Multithreading and Multiprocessing Tutorial.

But you have two distinct tasks: writing to a CSV file and sending data to an API. So it makes sense to split the code in two functions. If you will be doing parallel processing, then basically you would read the JSON one item at a time and feed a "queue". So you need some kind of function that accepts a URL, with optional payload and return whatever you want, that could be the just the response message based on the HTTP status code. Perhaps you can build something around existing code if you look at the link quoted above, and adapt it to your need.

I might edit this post later to add more suggestions on that.

I also recommend the book Expert Python Programming by Tarek Jaworski & Michal Ziade if you want to get more in depth knowledge on such topics.

\$\endgroup\$
1
  • \$\begingroup\$ Thanks for the message. I have created a function to write to CSV. But while calling the function I am getting I/O operation on the closed file. Any suggestions. P.S- I have edited the code \$\endgroup\$
    – Durga
    Jan 30, 2022 at 14:59

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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