norm/Gainsight
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Nothing left to do today, getting 500 errors for the enrollment and attempt resource endpoints. Everything is set up correctly for hitting those endpoints, but not sure if the 500 is on me or on the application/engineers.

Browse Source
This commit is contained in:
Norm Rasmussen
2024-09-01 16:49:30 -04:00
parent 35194352f5
commit 6323f577de
5 changed files with 196 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

146
Scripts/Migration_tool/Mark_Course_As_Complete/mark_course_as_complete.py
View File

@ -1,5 +1,5 @@
import Calls
import datetime
from datetime import datetime, timedelta
baseurl = Calls.BASEURL
@ -15,44 +15,55 @@ Order of operations:
8. Run migration.
"""
#
# class Project:
# proj_url = "projects"
#
# def __init__(self, PROJ_ID: str):
# self.PROJ_ID = PROJ_ID
#
#
# class Item:
# item_url = "items"
#
# def __init__(self, item_id: str):
# self.item_id = item_id
#
#
# class Courses:
# course_url = "courses"
#
# def __init__(self, course_uuid, course_name):
# self.course_uuid = course_uuid
# self.course_name = course_name
#
#
# class People:
# ppl_url = "people"
#
# def __init__(self, ppl_uuid, ppl_email):
# self.ppl_uuid = ppl_uuid
# self.ppl_email = ppl_email
class Project:
proj_url = "projects"
def __init__(self, proj_id: str):
self.proj_id = proj_id
class Item:
item_url = "items"
def __init__(self, item_id: str):
self.item_id = item_id
class Courses:
course_url = "courses"
def __init__(self, course_uuid, course_name):
self.course_uuid = course_uuid
self.course_name = course_name
class People:
ppl_url = "people"
def __init__(self, ppl_uuid, ppl_email):
self.ppl_uuid = ppl_uuid
self.ppl_email = ppl_email
PPL_UUID = ""
PPL_URL = "people"
PPL_EMAIL = ""
COURSE_URL = "courses"
COURSE_UUID = ""
COURSE_NAME = ""
ITEM_URL = "items"
ITEM_ID = ""
PROJ_URL = "projects"
PROJ_ID = ""
def get_people():
email = "norm@rsmsn.co"
if type(email) is str:
learner = get_individual_person(email)
global per
per = People(learner[0], learner[1])
get_courses(per.ppl_uuid)
PPL_UUID = learner[0]
PPL_EMAIL = learner[1]
get_courses(PPL_UUID)
elif type(email) is list:
person_uuids = get_group_person(email)
else:
@ -74,11 +85,10 @@ def get_individual_course(name, learner_uuid):
returned = Calls.get(url)
for items in returned["data"]:
global c
single_uuid = items["id"]
c = Courses(single_uuid, items["attributes"]["name"])
print(f"Cool. Course {single_uuid} exists. Checking enrollments.")
enrollment = get_enrollment_status(single_uuid, learner_uuid)
COURSE_UUID = items["id"]
COURSE_NAME = items["attributes"]["name"]
print(f"Cool. Course {COURSE_UUID} exists. Checking enrollments.")
enrollment = get_enrollment_status(COURSE_UUID, learner_uuid)
def get_enrollment_status(uuid, learner_uuid):
@ -86,46 +96,53 @@ def get_enrollment_status(uuid, learner_uuid):
f"{baseurl}/courses/{uuid}/enrollments?filter[person_id][eq]={learner_uuid}"
)
enrolled = Calls.get(enrollment_url)
if enrolled["data"] == "":
print(enrolled["data"])
if enrolled["data"] == []:
now = datetime.now()
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")
print("Oof, no enrollments. Not to worry. We'll get one created for you.")
mig_enroll_payload = {
"data": {
"attributes": {
"enrolled_at": {formatted_now},
"course_id": {c.course_uuid},
"person_id": {per.ppl_uuid},
{"enrolled_at": formatted_now},
{"course_id": COURSE_UUID},
{"person_id": PPL_UUID},
}
}
}
mig_enroll_url = Call.post(
f"{baseurl}migration/projects/{p.proj_id}/items/{i.item_id}/enrollment_resources"
mig_enroll_url = Calls.post(
f"{baseurl}migration/projects/{PROJ_ID}/items/{ITEM_ID}/enrollment_resources",
mig_enroll_payload,
)
print("Cool, enrollment resource has been created. Let's check that it exists.")
check_resources()
else:
print("Nice! We have an enrollment. So now we just need to update progress.")
create_attempt(enrolled["data"])
# create_attempt(enrolled["data"])
def create_attempt(data):
print(f"Is there a project ID? {p.proj_id}")
attempt_url = f"{baseurl}/migration/projects/{p.proj_id}/items/{i.item_id}/course_attempt_resources"
print(f"Is there a project ID? {PROJ_ID}")
attempt_url = f"{baseurl}/migration/projects/{PROJ_ID}/items/{ITEM_ID}/course_attempt_resources"
now = datetime.now()
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")
new_attempt_payload = {
"data": {
"attributes": {
"uuid": {"which uuid?"},
"display_name": {f"{per.ppl_email}'s Attempt for course {c.course_name}"},
"learner_id": {per.ppl_uuid},
"course_id": {c.course_uuid},
"display_name": {f"{PPL_EMAIL}'s Attempt for course {COURSE_NAME}"},
"learner_id": {PPL_UUID},
"course_id": {COURSE_UUID},
"progress": {"100"},
"started_at": {datetime.now() - timedelta(hours = 2)},
"started_at": {datetime.now() - timedelta(hours=2)},
"completed_at": {formatted_now},
"completed_activities": [{"uuid": {"1111"}, "completed_at": {formatted_now}}],
"completed_activities": [
{"uuid": {"1111"}, "completed_at": {formatted_now}}
],
}
}
}
check_resources()
"""
@ -136,15 +153,18 @@ def create_attempt(data):
)
"""
def check_resources():
print("Checking resources")
get_resources_url = (
f"{baseurl}/migration/projects/{p.proj_id}/items/{i.item_id}/resources"
f"{baseurl}/migration/projects/{PROJ_ID}/items/{ITEM_ID}/resources"
)
get_resources = Calls.get(get_resources_url)
if get_resources["data"] == []:
print("Something went wrong. No resources were created.")
else:
print("We're in! An enrollment exists, let's create an attempt.")
print("We're in! Here's the data:")
print(get_resources)
def create_project_item():
@ -156,24 +176,22 @@ def create_project_item():
"attributes": {"name": project_name},
}
}
proj_url = f"{baseurl}/migration/{Project.proj_url}"
print(proj_url)
tmp_p = Calls.post(proj_url, proj_payload)
global p
p = Project(tmp_p["data"]["id"])
print(f"Created Project Id: { p.proj_id }")
proj_full_url = f"{baseurl}/migration/{PROJ_URL}"
print(proj_full_url)
tmp_p = Calls.post(proj_full_url, proj_payload)
PROJ_ID = tmp_p["data"]["id"]
print(f"Created Project Id: {PROJ_ID}")
item_url = f"{baseurl}/migration/{p.proj_url}/{p.proj_id}/{Item.item_url}"
item_full_url = f"{baseurl}/migration/{PROJ_URL}/{PROJ_ID}/{ITEM_URL}"
item_payload = {
"data": {
"type": "migration_items",
"attributes": {"type": "courses"},
}
}
item_return = Calls.post(item_url, item_payload)
global i
i = Item(item_return['data']['id'])
print(f"Created Item ID: { i.item_id }")
item_return = Calls.post(item_full_url, item_payload)
ITEM_ID = item_return["data"]["id"]
print(f"Created Item ID: { ITEM_ID }")
def get_group_course(list):

94
Scripts/Migration_tool/migration_playground.py
View File

@ -18,6 +18,7 @@ Notes:
import json
from utils import calls, apikeys
from datetime import datetime, timedelta
JSONDOC = "./api_docs.json"
BASEURL = "https://api.northpass.com/v2/migration"
@ -35,7 +36,6 @@ tupee = [
# Get Resources (Done after you've added them from below)
# *************************
("/v2/migration/projects/{project_id}/items/{item_id}/resources", "get"),
# *************************
# Post calls for creating new resources.
# *************************
@ -129,13 +129,14 @@ def get_all_projects():
if ret["data"] == "":
for items in ret["data"]:
project_ids.append(items['id'])
project_ids.append(items["id"])
print(f"{ items['attributes']['name'] } -- { items['id'] }")
else:
print("Looks like there are no projects created!")
return project_ids
def delete_all_projects():
"""
Deletes all projects after returning the uuids of the projects currently available.
@ -145,6 +146,7 @@ def delete_all_projects():
url = f"{BASEURL}/projects/{proj}"
calls.delete(url)
def get_specific_project():
"""
Returns results from a specific project.
@ -154,9 +156,95 @@ def get_specific_project():
calls.get(url)
def create_enrollment():
project_id = "6c7a21c2-de35-4b9d-9b80-a235401783af"
item_id = "80b95e38-78d1-44b9-8d9f-be96d9c7bf6e"
learner_uuid = "101d891d-f145-4cb2-8f7f-f0d8a90a743e"
course_uuid = "0d41bb57-bc65-4e05-adfd-58436ed0bd50"
now = datetime.now()
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")
print("Creating Enrollment")
enroll_payload = {
"data": {
"attributes": {
"enrolled_at": formatted_now,
"course_id": course_uuid,
"person_id": learner_uuid,
}
}
}
enroll_url = calls.post(
f"{BASEURL}/projects/{project_id}/items/{item_id}/enrollment_resources",
enroll_payload,
)
print(f"{enroll_url} response is")
if enroll_url != 404:
print("Cool, enrollment resource has been created. Let's create an attempt.")
create_attempt(project_id, item_id, learner_uuid, course_uuid)
else:
print("A 404 error occurred.")
def create_attempt():
course_uuid = "f1b92092-60bd-4d52-922a-e462f132b69c"
project_id = "6c7a21c2-de35-4b9d-9b80-a235401783af"
item_id = "80b95e38-78d1-44b9-8d9f-be96d9c7bf6e"
learner_uuid = "101d891d-f145-4cb2-8f7f-f0d8a90a743e"
now = datetime.now()
formatted_now = now.strftime("%Y-%m-%d %H:%M:%S")
start_time = datetime.now() - timedelta(hours=2)
print(f"{formatted_now} ---- {start_time}")
new_attempt_payload = {
"data": {
"attributes": {
"uuid": course_uuid,
"display_name": f"norm@rsmsn.co's Attempt for course {course_uuid}",
"learner_id": learner_uuid,
"course_id": course_uuid,
"progress": "100",
"started_at": "2024-09-01 14:35:55",
"completed_at": formatted_now,
"completed_activities": [
{"uuid": "1111", "completed_at": formatted_now}
]
}
}
}
print(new_attempt_payload)
attempt_call = calls.post(
f"{BASEURL}/projects/{project_id}/items/{item_id}/course_attempt_resources",
new_attempt_payload,
)
print(attempt_call)
"""
If the learner is enrolled, we can go straight to creating an attempt.
(
"/v2/migration/projects/{project_id}/items/{item_id}/course_attempt_resources",
"post",
)
"""
def check_resources():
print("Checking resources")
get_resources_url = (
f"{baseurl}/migration/projects/{PROJ_ID}/items/{ITEM_ID}/resources"
)
get_resources = Calls.get(get_resources_url)
if get_resources["data"] == []:
print("Something went wrong. No resources were created.")
else:
print("We're in! Here's the data:")
print(get_resources)
if __name__ == "__main__":
# delete_all_projects()
# read_json_docs()
# create_project()
get_all_projects()
# get_all_projects()
# get_specific_project()
create_enrollment()
# create_attempt()

BIN
Scripts/Migration_tool/utils/__pycache__/__init__.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
Scripts/Migration_tool/utils/__pycache__/calls.cpython-310.pyc
View File

Binary file not shown.

36
Scripts/Migration_tool/utils/calls.py
View File

@ -1,10 +1,10 @@
import requests
import Apikeys
from utils import apikeys
import pprint
PP = pprint.PrettyPrinter(indent=4)
APIKEY = Apikeys.SANDBOX
APIKEY = apikeys.SANDBOX
HEADERS = {"content-type": "application/json", "X-Api-Key": APIKEY}
BASEURL = "https://api.northpass.com/v2"
@ -23,18 +23,28 @@ def get(url):
return json_get
# def post(url, payload):
# try:
# post_response = requests.post(url, headers=HEADERS, json=payload)
# print(f"Executed Post Request. Status code is {post_response.status_code}")
# except TypeError as h:
# print(
# f"Error occurred. Here's the info: {h} and status code: {post_response.status_code}"
# )
# finally:
# try:
# json_post = post_response.json()
# except JSONDecodeError as e:
# print(f"Error occurred. Here's the info: {e}.")
# finally:
# # PP.pprint(json_get)
# return json_post
#
def post(url, payload):
try:
post_response = requests.post(url, headers=HEADERS, json=payload)
print(f"Executed Post Request. Status code is {get_response.status_code}")
except HTTPError as h:
print(
f"Error occurred. Here's the info: {h} and status code: {get_response.status_code}"
)
finally:
json_post = get_response.json()
# PP.pprint(json_post)
post_response = requests.post(url, headers=HEADERS, json=payload)
print(f"Executed Post Request. Status code is {post_response.status_code}")
print(post_response.text)
return post_response
def delete(url):
try: