PowerClerk® API Integrations using Python can improve the efficiency of any PowerClerk program. Connecting digital systems to share data will reduce mismatches, errors, and improve accuracy. Automating review processes, updating results from inspections, or work order statuses can reduce laborious human interactions. This can free up valuable time for managers, engineers and analysts dealing with your program backlog.
Python is a popular programming language often used as glue to connect systems together. As well as being one of the easier programming languages to learn, many systems (e.g., CYME, Safe FME for ArcGIS, Salesforce, PSS/E) provide packages for Python integrations or include an interpreter to run Python directly in the application.
The PowerClerk API enables connected clients to perform almost any operation an applicant or administrator can do manually with the PowerClerk UI. If you are not familiar with the PowerClerk API—particularly the new features of V2 like OAuth 2.0 and JSON support—documentation is available at: https://apidocs.powerclerk.com.
The rest of this article walks through well-tested Python code to integrate with the PowerClerk API. The first five steps describe a Python class that you can use as a template to avoid common pitfalls. The examples at the end show you how easy it is to use!
As is typical for these types of articles, however, a standard disclaimer: The code examples below are provided for illustrative purposes only, not officially supported by Clean Power Research, and is provided “as is” without warranty of any kind.
Step one: Import the packages we need
The Python requests library is the recommended way if calling HTTP REST endpoints like the PowerClerk API. This is the only package we’ll use that is not included in Python standard library.
We need the base64 module to encode our OAuth 2.0 client credentials. We’ll use the time module to ensure we don’t exceed the API rate limit and monitor token expiration. The os module is a suggested way to get PowerClerk credentials from environment variables, more secure than storing them in code.
import requests
import base64
import time
import os
Step two: Configure constants
We need an API Key, Client Id and Secret for secure access to PowerClerk. If you don’t have these, our documentation has instructions to help you get them.
The URLs in this example are for the PCI Trial sandbox and would need to be updated for production or another PowerClerk test instance.
Finally, MAX_TPM is the API key rate limit in transactions per minute and set to the default of 60. Exceeding the rate limit can result in HTTP 429 “Too Many Requests” errors from PowerClerk. So, if you share your API key, or plan to run multiple threads for your integration, you can reduce this value accordingly.
# Set your PowerClerk API key and credentials
API_KEY = os.getenv('POWERCLERK_API_KEY') or '<YOUR API KEY>'
CLIENT_ID = os.getenv('POWERCLERK_CLIENT_ID') or '<YOUR CLIENT ID>'
SECRET = os.getenv('POWERCLERK_SECRET') or '<YOUR CLIENT SECRET>'
API_URL = 'https://api.cleanpowerdemo.com/PCITrial/services/v2'
IDP_URL = 'https://identity.cleanpowerdemo.com/PCITrial/connect/token'
MAX_TPM = 60 # Should be 60 or fewerStep three: PowerClerk class
To keep things tidy, we’ll build a PowerClerk class. We can initialize it with our constants, the encoded authorization string ready to fetch access tokens, and variables for throttling requests.
class PowerClerk(object):
# Initialize the PowerClerk object
def __init__(self,
api_root=API_URL, idp_root=IDP_URL,
api_key=API_KEY,
client_id=CLIENT_ID, secret=SECRET,
max_tpm=MAX_TPM,
) -> None:
# Store the initialization variables
self.api_root = api_root
self.idp_root = idp_root
self.api_key = api_key
# Variables for OAuth 2.0
self.__oauth_basic = base64.b64encode(bytes(f'{client_id}:{secret}', 'utf-8')).decode()
self.__token = {}
# Variables for throttling
self.__tick_duration = 1 / (max_tpm / 60)
self.__last_api_call_time = time.time() - 60
Step four: Request throttling
This method will make sure we don’t go over PowerClerk’s API rate limit and cause unnecessary errors.
The __tick_duration variable was initialized to the minimum time between calls, and we set the __last_api_call_time to the current time when making an API call. So all we need to do is pause (sleep) between calls if there was not already enough of a delay.
def __throttle_api_calls(self):
time_since_last_call = time.time() - self.__last_api_call_time
if time_since_last_call < self.__tick_duration:
time.sleep(self.__tick_duration - time_since_last_call)
self.__last_api_call_time = time.time()
The PowerClerk API V2 uses the OAuth 2.0 client credentials flow. This method ensures we always use a fresh access token by fetching a new one if the current token has expired.
def __get_oauth_token(self):
time_now = time.time()
# If there is no token, or it expired, fetch a new token
if not self.__token or self.__token['exp'] < time_now:
headers = {
'Authorization': f'Basic {self.__oauth_basic}',
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json',
}
response = requests.post(self.idp_root,
headers=headers,
data={'grant_type': 'client_credentials'})
if not response.ok:
raise RuntimeError(f'Authentication failed with HTTP status {response.status_code} : {response.text}')
body = response.json()
access_token = body.get('access_token')
expires_in = body.get('expires_in')
self.__token = {
'exp': time_now + expires_in - 60,
'token': access_token
}
return {'Authorization': f'Bearer {self.__token["token"]}'}
Step six: Request methods
Finally, we need methods to make the API calls. The get and post methods are wrappers for the request method where all the action happens.
First, we call the method to keep us below the rate limit. Then we build the header for the PowerClerk API including the credentials. This assumes we want to use JSON as the API payload (you’re using Python, so why wouldn’t you!) but it would be easy to change if you prefer XML.
def request(self, path, method, **kwargs):
self.__throttle_api_calls()
headers = kwargs.pop('headers', {}) | self.__get_oauth_token() | {
'X-ApiKey': self.api_key,
'Accept': 'application/json',
}
response = requests.request(method, self.api_root + path, headers=headers, **kwargs)
if not response.ok:
raise RuntimeError(f'API call failed with HTTP status {response.status_code}: {response.text}')
return response
# Convenience method for GET calls. Returns the requests response
def get(self, path, **kwargs):
return self.request(path, 'GET', **kwargs)
# Convenience method for POST calls. Returns the requests response
def post(self, path, **kwargs):
return self.request(path, 'POST', **kwargs)
Examples
The PowerClerk class is boilerplate code that makes the actual work with the PowerClerk API quite simple.
Assuming we’re happy with the default parameters from step two and our credentials are nicely tucked away as environment variables (or you just hard-coded them and promise to come back and fix it later!) then it could not be simpler to create the PowerClerk object.
pc = PowerClerk()
First, we’ll set up some constants for the Program Id and Project Number. The Form Id can be a Public Id or a Custom Id. Again, replace these example values with your own.
Now we use the get method to call the Get Project Data API. The request just needs the API path and returns a requests response object where we can extract the payload.
PROGRAM_ID = 'M5JGHPZK7GFC' # Unique for your program and found in the PowerClerk URL
PROJECT_NUMBER = 'INTG-00256' # A project we have access to
FORM_ID = 'TECHNICAL_REVIEW' # Example of a Custom Id mapped on the Custom API Ids page
response = pc.get(f'/Programs/{PROGRAM_ID}/Projects/{PROJECT_NUMBER}/Forms/{FORM_ID}/Data')
print(response.json())
A request to the Set Project Data API is almost as simple using the post method with a payload. Here we build an example payload simulating Hosting Capacity Analysis results using two data fields with Custom Ids.
hca_result = {'DataFieldId': 'HCA_RESULTS', 'Value': 'Pass'}
hca_engineer = {'DataFieldId': 'HCA_ENGINEER', 'Value': 'Python powered!'}
new_project_data = {'Project': {'DataFields': [hca_result, hca_engineer]}}
response = pc.post(f'/Programs/{PROGRAM_ID}/Projects/{PROJECT_NUMBER}/Forms/{FORM_ID}/Draft/Data', json=new_project_data)
print(response.json())
That data was set as Draft—equivalent to a user updating a form before submitting it—so we need to make a final call to the Submit API.
response = pc.post(f'/Programs/{PROGRAM_ID}/Projects/{PROJECT_NUMBER}/Forms/{FORM_ID}/Draft/Submit')
print(response.json())
Bonus attachment examples!
The examples we have seen so far use JSON payloads but it’s possible to use binary formats for attachments.
Again, we’ll use a Custom Id for the Attachment Id. We can download attachments and even extract the filename from the Content-Disposition header before saving them to a file.
ATTACHMENT_ID = 'SINGLE_LINE_DIAGRAM'
response = pc.get(f'/Programs/{PROGRAM_ID}/Projects/{PROJECT_NUMBER}/Forms/{FORM_ID}/Attachments/{ATTACHMENT_ID}/Download')
print(f'Downloaded {len(response.content)} byte attachment')
filename = response.headers['Content-Disposition'].split('"')[1]
with open(filename, "wb") as f:
f.write(response.content)
To upload an attachment, we need to include the content headers in the request. Again, this is uploaded as Draft, so a Submit API call is needed to store it.
FILENAME = 'single_line_diagram.pdf'
with open(FILENAME, "rb") as f:
attachment = f.read()
headers = {
'Content-Type': 'application/pdf',
'Content-Disposition': f'attachment; filename="{FILENAME}"'
}
response = pc.post(
f'/Programs/{PROGRAM_ID}/Projects/{PROJECT_NUMBER}/Forms/{FORM_ID}/Draft/Attachments/{ATTACHMENT_ID}/Upload',
headers=headers,
data=attachment)
Let us know if you find this technical article useful and we’ll write more. Please reach out if you have questions, suggestions, or just to discuss what you plan to build.