'How to iterate sample users along with endpoint & payload while calling each API post request
Original question link: There are two functions:
ReadInp() -> Dict[str, str]
This reads a csv and returns a dict with a url as key and a payload as the value.ReadTUsr() -> List[List[str, str]]
This reads a separate csv and returns 'sample users' as a list of lists. Each inner list is an id type and an id value. For instance:
[['uId', 'T1'], ['srId', 'T2'], ['siId', 'T3'], ['siId', 'T4'], ['srId', 'T5']...]
I need to make url calls that join these two values. The url being called will come from the first return value, but it must splice in the sample user from the second list. These lists may not be the same size, and if the second is exhausted before the first it should iterate again over the list until the first is exhausted.
For instance, the resultant calls would be like:
http://localhost:8080/cabApp/callcab?uId=T1
then POST body:
{'location': '', 'ppl': '2', 'text': 'MyCab'}
Or:
http://localhost:8080/cabApp/call3cab?srId=T2
then POST body: {'display': 'RMP', 'time': '', 'location': 'LA'}
This is what I have tried: Trial 1
def makeAcall():
r = ReadInp() # gives dict with endpoint as key and payload as Value
# This dict looks like this >> {'cabApp/callcab': {'location': '', 'ppl': '2', 'text': 'MyCab'}, 'cabApp/call3cab': {'display': 'RMP', 'time': '', 'location': 'LA'}, ...}
t = ReadTUsr() # give nested user list
# This list looks like this >> [['uId', 'T1'], ['srId', 'T2'], ['siId', 'T3'], ['siId', 'T4'], ['srId', 'T5']...]
print(t)
print(type(r))
#for k1,v1 in r.items():
# for i,j in t:
for (k1,v1), (i,j) in zip(r.items(), t):
url = f"{AURL}(k1)}"
rj = v1
#uid = {"uId": "T1"}
headers = {'Accept': "application/json"}
res = requests.post(url, json=rj, headers=headers, params=(i,j))
print("Req Resp Code:",res.status_code)
#api_req_res_text.append(res.text)
api_req_res_code.append(res.status_code)
return api_req_res_code
Trial 2
def equalize_lists(users: list, desired_length) -> list:
if len(users) > desired_length:
return users[:desired_length]
else: # We need to extend the user list
#repeat, extra = (desired_length // len(users), desired_length % len(users))
repeat, extra = divmod(desired_length, len(users))
users *= repeat + (extra > 0)
#for _ in range(repeat):
#users.extend(users)
#users.extend(users[:extra])
return users
def makeAcall():
r = ReadInp() # gives dict with endpoint as key and payload as Value
# This dict looks like this >> {'cabApp/callcab': {'location': '', 'ppl': '2', 'text': 'MyCab'}, 'cabApp/call3cab': {'display': 'RMP', 'time': '', 'location': 'LA'}, ...}
t = ReadTUsr() # give nested user list
# This list looks like this >> [['uId', 'T1'], ['srId', 'T2'], ['siId', 'T3'], ['siId', 'T4'], ['srId', 'T5']...]
print(t)
print(type(r))
users = equalize_lists(t, len(r))
#users *= repeat + (extra > 0)
for ((k1, v1), (i, j)) in (zip(r.items(), users)):
url = f"{AURL}{k1}"
rj = v1
headers = {'Accept': "application/json"}
res = requests.post(url, json=rj, headers=headers, params=(i,j))
#uid = cycle(t)
print("Req Resp Code:",res.status_code)
#api_req_res_text.append(res.text)
api_req_res_code.append(res.status_code)
return api_req_res_code
- Note1: This is how my dict looks like
{'cabApp/callcab1': {'location': '', 'ppl': '2', 'text': 'MyCab'}, 'cab2App/call2cab2': {'ppl': '', 'text': 'Cab2'}, 'cabApp/call3cab': {'display': 'RMP', 'time': '', 'location': 'LA'}...}
Here cabApp/callcab1, cab2App/call2cab2 & cabApp/call3cab are keys, which are endpoints[iterated as k1 in code] and rest all is their individual payloads [iterated as v1] respectively. http://localhost:8080 part is stored in AURL and is static. I can not update the code which builds the API as per the instruction[res where I am making the post request]. What I am searching for is the way to iterate list and use it as value after param. - Note2: We can change the way it read sample users csv. Right now I am using this code to read sample users csv which gives nested list.
def ReadTUsr():
with open(T_U_csv, 'r') as file:
csv_file = csv.reader(file)
data = [it for it in csv_file]
#print(data)
return data
- Note3: Earlier ReadTUsr() was dict and was working as expected (was able to iterate and call new sample user for each request) but latest code has duplicate keys in dict (multiple times siId & srId) so it only takes one record from duplicate keys which is default python behavior.
- Que How to iterate sample users (i,j) along with endpoint & payload (k1,v1) while calling each post request for res variable.
Solution 1:[1]
def equalize_lists(users: list, desired_length) -> list:
if len(users) > desired_length:
return users[:desired_length]
else: # We need to extend the user list
#repeat, extra = (desired_length // len(users), desired_length % len(users))
repeat, extra = divmod(desired_length, len(users))
users *= repeat + (extra > 0)
#for _ in range(repeat):
#users.extend(users)
#users.extend(users[:extra])
return users
def makeAcall():
r = ReadInp() # gives dict with endpoint as key and payload as Value
# This dict looks like this >> {'cabApp/callcab': {'location': '', 'ppl': '2', 'text': 'MyCab'}, 'cabApp/call3cab': {'display': 'RMP', 'time': '', 'location': 'LA'}, ...}
t = ReadTUsr() # give nested user list
# This list looks like this >> [['uId', 'T1'], ['srId', 'T2'], ['siId', 'T3'], ['siId', 'T4'], ['srId', 'T5']...]
print(t)
print(type(r))
users = equalize_lists(t, len(r))
# or use this >> users = itertools.cycle(ReadInp())
#users *= repeat + (extra > 0)
for ((k1, v1), (i, j)) in (zip(r.items(), users)):
url = f"{AURL}{k1}"
rj = v1
headers = {'Accept': "application/json"}
res = requests.post(url, json=rj, headers=headers, params={i,j})
#uid = cycle(t)
print("Req Resp Code:",res.status_code)
#api_req_res_text.append(res.text)
api_req_res_code.append(res.status_code)
return api_req_res_code
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
Solution | Source |
---|---|
Solution 1 | Maddy9 |