Walmart new script for muiltiple ids (activities, learning paths, courses). Talkspace script to update their workato recipe.

This commit is contained in:
Norm Rasmussen
2023-10-17 17:28:07 -04:00
parent c4e8f9ae7f
commit 7450785c11
6 changed files with 326 additions and 23 deletions

View File

@ -0,0 +1,105 @@
import csv
import os
import Apikeys
import requests
APIKEY = Apikeys.walmartprod
CMD = "touch ~/Downloads/Spark_Categories.csv"
HEADERS = {
"accept": "application/json",
"X-Api-Key": APIKEY,
}
def get_categories():
"""
Function to get the category Name and IDs and adding them to a list of tuples
"""
os.system("touch ~/Downloads/Spark_Categories.csv")
url = "https://api2.northpass.com/v2/categories?limit=100"
response = requests.get(url, headers=HEADERS)
response = response.json()
for item in response["data"]:
cat_tupe = (item["id"], item["attributes"]["name"])
with open(
"/Users/normrasmussen/Downloads/Spark_Categories.csv", "a", newline=""
) as csvfile:
writer = csv.writer(
csvfile, delimiter=",", quotechar="|", quoting=csv.QUOTE_MINIMAL
)
writer.writerow(cat_tupe)
def get_courses():
"""
Function to get the course name, ids, and activity names and ids to a list.
"""
count = 0
while True:
count += 1
url = f"https://api2.northpass.com/v2/courses?page={count}"
response = requests.get(url, headers=HEADERS)
response = response.json()
nextlink = response["links"]
for item in response["data"]:
status = item["attributes"]["status"]
if status == "live":
courses = (item["id"], item["attributes"]["name"])
get_activities(courses)
if "next" not in nextlink:
break
def get_activities(courses):
"""
Function that exists within a for loop!
This function is to get the category Name and IDs and adding them to a list of tuples
and writes them to a csv
"""
os.system("touch ~/Downloads/Spark_Courses.csv")
course_list = []
course_uuid, _ = courses
url = f"https://api2.northpass.com/v2/courses/{course_uuid}/activities"
response = requests.get(url, headers=HEADERS)
response = response.json()
for data in response["data"]:
activity_tupe = (data["id"], data["attributes"]["title"])
course_list.append(activity_tupe)
courses = (courses, course_list)
with open("/Users/normrasmussen/Downloads/Spark_Courses.csv", "a") as csvfile:
writer = csv.writer(
csvfile, delimiter=",", quotechar="|", quoting=csv.QUOTE_MINIMAL
)
writer.writerow(courses)
def get_learning_paths():
"""
Function to get the Learning Path Name and IDs and adding them to a list of tuples
"""
os.system("touch ~/Downloads/Spark_LPs.csv")
url = "https://api2.northpass.com/v2/learning_paths"
response = requests.get(url, headers=HEADERS)
response = response.json()
for data in response["data"]:
lps = (data["id"], data["attributes"]["name"])
print(lps)
with open("/Users/normrasmussen/Downloads/Spark_LPs.csv", "a") as csvfile:
writer = csv.writer(
csvfile, delimiter=",", quotechar="|", quoting=csv.QUOTE_MINIMAL
)
writer.writerow(lps)
if __name__ == "__main__":
get_categories()
get_courses()
get_learning_paths()

View File

@ -1,8 +1,7 @@
import requests
import Apikeys
import csv
import time
import os
import Apikeys
import requests
apikey = Apikeys.walmartprod
cmd = "touch ~/Downloads/Spark_Categories.csv"

View File

@ -6,8 +6,11 @@ import requests
import pandas as pd
import Apikeys
APIKEY = Apikeys.walmartprod
APIKEY = Apikeys.talkspace_1099
COURSES = ["Getting Started at Talkspace (ICP - 2023)",
"Getting Started at Talkspace (Prescriber)",
"Getting Started at Talkspace (Associate ICP)",
"Talkspace Managed Care Plans"]
def get_course():
"""
@ -20,8 +23,7 @@ def get_course():
while True:
count += 1
url = f"https://api2.northpass.com/v2/courses?page={count}"
print(url)
url = f"https://api.northpass.com/v2/courses?page={count}"
headers = {"accept": "application/json", "X-Api-Key": APIKEY}
response = requests.get(url, headers=headers)
data = response.json()
@ -32,22 +34,25 @@ def get_course():
if status == "live":
uuid = response["id"]
name = response["attributes"]["name"]
full_description = response["attributes"]["full_description"]
course_dict = {
"id": uuid,
"name": name,
"status": status,
"full_description": full_description,
}
if name in COURSES:
# full_description = response["attributes"]["full_description"]
course_dict = {
"id": uuid,
"name": name,
# "status": status,
# "full_description": full_description,
}
try:
courses.append(course_dict)
except TypeError as e:
print(f"Error: {e}")
finally:
write_to_csv(courses)
else:
pass
try:
courses.append(course_dict)
except TypeError as e:
print(f"Error: {e}")
finally:
# write_to_csv(courses)
pass
else:
pass
print(courses)
if "next" not in nextlink:
break