Prequisites : Setup Connected App with X509 Certification.
預設需要先在Salesforce 上載X509證書。詳細設定連接。(可接受自行簽署版本證書)
Details Set up in Salesforce side : link
#POC of JWT call to API#Colab install library with >> !pip install pyJWT[crypto]
import jwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
import time
import requests
import json
expTime = int(time.time()) + 900 #Unix Epoch Timestamp Expire time. 15mins buffer
jwt_algorithm = 'RS256'
jwt_private_key_file = '/myPath/server.key'
jwt_payload = {
"iss": "clientid",
"sub": "salesforce_username_email",
"aud": "https://test.salesforce.com",
"exp": expTime
}
jwt_header = {
"alg": "RS256",
"typ": "JWT"
}
salesforceEndpoint = 'https://test.salesforce.com/services/oauth2/token'
api_endpoint = 'https://instance_domain.my.salesforce.com/services/data/v57.0/sobjects/myObject__c'
test_data={
"description": "PythonSent",
"language": "en",
"remarks": "xxx",
"expiry_time": expTime+900,
"created_time": expTime
}
def loginSfdc():
# Load the private key
with open(jwt_private_key_file, 'r') as f:
jwt_private_key = f.read()
# Generate the JWT token
jwt_token = jwt.encode(jwt_payload, jwt_private_key, algorithm=jwt_algorithm,headers=jwt_header)
print('token : ' + jwt_token);
salesforceUrlParams = {
'grant_type' : 'urn:ietf:params:oauth:grant-type:jwt-bearer',
'assertion' :jwt_token
}
response = requests.post(salesforceEndpoint, params=salesforceUrlParams, verify=False)
if response.status_code == 200:
authToken = response.json()['access_token']
print('return token ' + authToken)
return authToken
else:
print('Request failed with status code:', response.status_code)
print('Request failed with status code:', response.text)
return(0)
#function to call
def callSfdc_pocApi(authToken):
session = requests.Session()
rheaders = {
'Authorization': 'Bearer ' + authToken,
'Content-Type': 'application/json'
}
response = requests.post(api_endpoint,headers=rheaders, data=json.dumps(test_data))
returnStr = ''
if response.status_code == 200:
data = response.json()
returnStr = data
else:
print('Request failed with status code:', response.text)
returnStr = response.status_code
return returnStr
#Main
response1 = loginSfdc()
if not response1 == 0:
response2 = callSfdc_pocApi(response1)
print ("OK")
print(response2)
else:
print ('Failed to Login')