import pandas as pd import requests import time """ Note that all single-line hash comments are referencing the code _below_ the comment line. """ APIKEY = "INSERT API KEY BETWEEN QUOTES" HEADERS = { "accept": "application/json", "X-Api-Key": APIKEY, } BASEURL = "https://api.northpass.com/v2/" IMPORTFILE = "INSERT THE PATH TO THE CSV ON YOUR LOCAL MACHINE" def bulk_invite_and_group(): """ Bulk endpoint which invites new people and adds them to a group via this structure: { "email": "me@mac.com" "groups": "GroupA" } This function looks for the group in the CSV as well. """ # This first line reads the CSV using pandas and puts the dataframe into a variable data = pd.read_csv(IMPORTFILE) # Grab all unique values from the Group column. It should only be 3 values (the 3 groups you're adding members to) groups = data["Group"].unique() # Turn it into a list so we can loop through it. groups = list(groups) # Outputting status updates for your reference print("Here are all groups within the CSV:") print(groups) print(" ") # Now we loop through the groups for group in groups: payload = "" # You can comment this out if you want. This just confirms the loops is moving through all the groups. print(group) # Creating a temporary variable for one group at a time tmp_group = data[data.Group == group] # For that group, grab all the emails using the column called "Email" people = list(tmp_group["Email"]) # Making sure the emails don't have commas or extra spaces. group = str(tmp_group["Group"].unique())[2:-2] # Show you how many people are in that group! We're making progress! print(f"Group --> {group} ... Amount of People --> {len(people)}") # Now let's prepare the actual API call. url = f"{BASEURL}bulk/people" # The reason we are checking for length is because our bulk endpoints can't take more than 1500 parameters at a time. # So if the amount of people in this group (remember, we're still in the for loop!) is greater than 1500, we'll have to break it up if len(people) > 1500: # Take 1500 people at a time, and add them to a mini payload (called mini load) and then we'll insert that into the bigger payload. for chunk in range(0, len(people), 1500): i = chunk payload_1 = [] i_to_add = people[i : i + 1500] for person in i_to_add: miniload = {"email": person, "groups": group} payload_1.append(miniload) # This print statement should show 1500 people. print(f"The long length {group} payload has {len(payload_1)}") # Here's where we add the mini-payload list to the formatted payload that the API needs payload = {"data": {"attributes": {"people": payload_1}}} # And here we go! Making the call to our endpoint response = requests.post(url, headers=HEADERS, json=payload) # Printing the status code for your awareness print(f"Completed. Status code is {response.status_code}") # What if the amount of all the people per group is less than 1500? Well, let's run this code block/else statement! else: payload_1 = [] # Pretty straight forward, we don't need to do any chunking, so just add everyone to the mini-payload, as above. for person in people: miniload = {"email": person, "groups": group} payload_1.append(miniload) # Same print statement as above for continuity. print(f"The {group} payload has {len(payload_1)}") # Constructing the payload the API endpoint expects! payload = {"data": {"attributes": {"people": payload_1}}} # And here we go again! Pushing the data to your academy. response = requests.post(url, headers=HEADERS, json=payload) # Showing you the status code - you should expect 200. print(f"Completed. Status code is {response.status_code}") print(response.text) # Next, we need to add all the properties (the Agency Name) to each person. # The reason why we use time.sleep() is to just give the application a few minutes to fully add everyone. # It's not really needed, but better to be safe than sorry print("Running add props from func...") time.sleep(3) add_props_from_func(people, data, group) def add_props_from_func(people, data, group): # Creating an empty list of emails that cause errors, to be filled if an error arises. errorlist = [] # Move through all the people for learner_email in people: # Grab the Agency name from the original dataframe, looking for the column called "AgencyName" agency_name = data.loc[data["Email"] == learner_email, "AgencyName"] # Clean up white space and commas agname = str(agency_name.values)[2:-2] # Print out confirmation that the email, agency, and group all align. print(f"Learner: {learner_email} --> Agency: {agname} from Group: {group}") # We first need the learner's UUID since that isn't passed in the CSV and is required for the properties endpoint. ppl_search = f"{BASEURL}people?filter[email][eq]={learner_email}" ppl_response = requests.get(ppl_search, headers=HEADERS) # Let's give it a shot to see if the person exists (they should, we just added them!) # If they exist, grab the ID and go straight into adding their properties via a different endpoint. try: ppl_data = ppl_response.json() learner_uuid = ppl_data["data"][0]["id"] # Now update the props prop_url = f"{BASEURL}properties/people/bulk" payload = { "data": [ { "attributes": {"properties": {"agency_name": agname}}, "id": learner_uuid, "type": "person_properties", } ] } # This is the actual call for adding the property to the person propresponse = requests.post(prop_url, json=payload, headers=HEADERS) # However, if we don't get a 200 status code, let's add the person to the error list for later investigation. if propresponse.status_code != 200: error_tupe = (learner_uuid, learner_email, agname) errorlist.append(error_tupe) # But, if the status code is 200, then we're good! And we output that it was successful. else: print(f"Looks like {learner_email} and {agname} was successful. Received status code: {propresponse.status_code}.") except (TypeError, IndexError) as e: error_tupe = (0, learner_email, agency_name) errorlist.append(error_tupe) print(f"{e} has occurred with {learner_email}") finally: pass # As one last print, show the error list. It should be empty! print(f"Error list: {errorlist}") if __name__ == "__main__": bulk_invite_and_group()