Accessing SKY API via Python for Database Export

Jax Murphy
Jax Murphy
Kudos 1 Name Dropper Facilitator 1
edited September 25 in SKY Developer General Discussions

Hello,

After posting here about my goal of Mirroring RE NXT Data for Use in Power BI, I read a lot of excellent discussion and received some great ideas. While Power Automate seemed like the most versatile solution, I was trying to find a solution that didn't require any financial investment.

One suggestion was to make use of the Query API via a script. I'm not a strong programmer and had never worked with an API prior to this. I learned a ton along the way, but was a little frustrated with the lack of documentation or discussion surrounding this approach. Eventually, largely through trial and error, I got a Python script to do everything I wanted. I have decided to share my journey here so that others may use this solution in the future, should they be interested (or opposed to Power Automate).

Before starting these instructions, make sure you have an IDE that supports Python and a SQL Server instance with a database and tables that you can write to. I recommend creating a dedicated user for the database so that you're not plugging your credentials into this script.

  1. Create an app in Blackbaud
    1. Go to My Applications
    2. Select New application and fill out the info (I set the URL to my org's homepage)
    3. Set the redirect URI to http://localhost:8080
    4. Add any contributors
    5. Note down the client_id and client_secret
  2. Write or identify a query in RE NXT
    1. Write the queries to extract the data you're looking for in RE NXT
    2. Copy down and edit the first Python script
    3. In the same root folder as the script, create a sky_credentials.json file and edit it to look like this:
      {
      "client_id": "      your_client_id",



      "client_secret": "      your_client_secret",
      "redirect_uri": "http://localhost:8080"
      }
    4. In the same root folder, create a txt file, name it .env and input your Blackbaud API Key, your client_id, and your client_secret like this:

      BB_API_KEY="your_api_key"
      CLIENT_ID="your_client_id"
      CLIENT_SECRET="your_client_secret"
    5. Run the first Python script, using the Try-It editor in the API reference to help construct the requisite URL, to identify the query you're looking for based on search parameters. For example, to search for a query titled "PowerBI Gifts", I would use the URL "https://api.sky.blackbaud.com/query/queries?product=RE&module=None&limit=200&search_text=PowerBI%Gifts"
      Note: The first time the script runs, it will open a web browser to authenticate. After this, it caches your token for future use and should not require authentication unless the .sky-token file is deleted.
    6. Identify the query and note down its ID. As far as I know, this is the ONLY way to retrieve the ID of a query, as the information is not available in the query editor in RE NXT
  3. Execute the query, poll job status until complete, and write to database
    1. Run the second Python script after editing to include the correct URL, API Subscription key, query ID, and SQL credentials

First Python Script:

from sky import Sky
from dotenv import load_dotenv
import urllib.request, json


url = 'your_url_from_try-it_editor'
load_dotenv()


client = Sky(
    api_key=None,
    file_path='sky_credentials.json',
    token_path=".sky-token"
)
_ = client.getLevels()
client_dict = client._loadCachedToken()["access_token"]
#print(client_dict)

try:

    hdr = {
        # Request headers
        'Content-Type': 'application/json',
        'Cache-Control': 'no-cache',
        'Bb-Api-Subscription-Key': 'your_subscription_key',
        'Authorization': 'Bearer ' + client_dict
    }

    req = urllib.request.Request(url, headers=hdr)

    req.get_method = lambda: 'GET'
    response = urllib.request.urlopen(req)
    print('Success')
    json_obj = json.loads(response.read().decode('utf-8'))
    #print(json_obj)
    #print(json_obj["queries"])
    new_obj = {index: value for index, value in enumerate(json_obj["queries"])}
    #print(new_obj)
    for i in new_obj.values():
        print(i)

except Exception as e:

    print('Error ', e)

Second Python script:

from sky import Sky
from dotenv import load_dotenv
import urllib.request, json, time
import pandas as pd
from sqlalchemy import create_engine

'''
Initialization
'''

#Load environment variables
load_dotenv()

#Define client variable as Sky class, passing credentials from file and defining token cache location
client = Sky(
    api_key=None,
    file_path='sky_credentials.json',
    token_path=".sky-token"
  )

#Make an HTTPS call to initialize token retrieval and/or refresh token. Ignore the invalid request error message in the console, we just need to run a call
_ = client.getLevels()

#Identify session access token
client_dict = client._loadCachedToken()["access_token"]

'''
First Query
'''

#Requests to run an existing query based on its ID
try:
    print('Attempting to submit query...')
    url = "https://api.sky.blackbaud.com/query/queries/executebyid?product=RE&module=None"
    hdr = {
        # Request headers
        'Content-Type': 'application/json',
        'Cache-Control': 'no-cache',
        'Bb-Api-Subscription-Key': 'your_subscription_key', #Our Sky API Subscription key
        'Authorization': 'Bearer ' + client_dict    #Passes the access token
    }

    # Request body

    data = {
    "id": [query_id], #the ID of our query
    "ux_mode" : "Synchronous", #Synchronous because we are waiting for the job to complete

}

    data = json.dumps(data) #Converts 'data' dictionary to json file

    #Make our POST request
    req = urllib.request.Request(url, headers=hdr, data=bytes(data.encode("utf-8")))
    req.get_method = lambda: 'POST'
    response = urllib.request.urlopen(req)


    #Convert HTTPS Response into readable dictionary
    json_dict = response.read().decode('utf-8') #Decodes byte-type HTTPS response to utf-8
    new_dict = json.loads(json_dict)    #Loads json back into a dictionary
    print('Query submitted successfully. Beginning status check and write.')   #Indicates we did not hit an error when making our request
except Exception as e:

    print('Error ',e)   #Prints the HTTPS error code response upon failure


'''
Second Query
'''

#Checks status of previous request and waits for completion, then writes the output to a database
try:
    #Identify job by ID, given by the new_dict
    new_url = "https://api.sky.blackbaud.com/query/jobs/"+new_dict["id"]+"?product=RE&module=None&include_read_url=OnceCompleted"

    hdr ={
        'Content-Type': 'application/json',
        'Cache-Control': 'no-cache',
        'Bb-Api-Subscription-Key': 'your_subscription_key',
        'Authorization': 'Bearer ' + client_dict
    }

    #Run a GET request
    req = urllib.request.Request(new_url, headers=hdr)
    req.get_method = lambda: 'GET'
    response = urllib.request.urlopen(req)

    #Convert to utf-8 again
    convert = response.read().decode('utf-8')
    new_json_obj = json.loads(convert)


    #Wait until status is set to Completed, checking on it every 30 seconds
    print("Waiting...")

    try:
        while(new_json_obj["status"] != 'Completed'):
            response = urllib.request.urlopen(req)
            convert = response.read().decode('utf-8')
            new_json_obj = json.loads(convert)
            time.sleep(30)
        print("Success")
    except Exception as e:
        print('Error ',e)


    #Extract the URL for scraping
    download = new_json_obj['sas_uri']


    #Read data into pandas dataframe
    df = pd.read_csv(download, encoding='latin-1', low_memory=False)
    print(df)
    #Store SQL Server credentials
    server = 'server_name'
    database = 'database'
    username = 'username'
    password = 'password'

    connection_string = (
        f"mssql+pyodbc://{username}:{password}@{server}/{database}"
        "?driver=ODBC+Driver+17+for+SQL+Server"
    )

    engine = create_engine(connection_string)


    #Write to SQL Server
    df.to_sql(name='Gifts', con=engine, if_exists='replace', index=False)

except Exception as e:
    print(e)

Hopefully this is helpful to anyone looking to make use of the Query API via Python. This was a rather quick write-up and may be lacking some details. I can share more information and answer related questions. Give it a try, noting that anything in bold is meant to be replaced by your own value. Also note that the second script can take upwards of 10 minutes to run depending on the size of your query's result set.

Tagged:

Answers

Categories