keycloak 获取 access token validate token

import requests as req
import json

reqJson = {
    "client_id":"jwt-service",
    "client_secret": "qaaaaa-468a-4ba6-b71a-21672d1376be",
    "username": "aaa@aaa.com",
    "password": "123123",
    "grant_type": "password"
}

tokenUrl = 'http://localhost:8080/auth/realms/dev/protocol/openid-connect/token'

header = {
    "Content-Type": "application/x-www-form-urlencoded",
} 

response = req.post(tokenUrl, data=reqJson, headers=header)
if response.status_code > 200:
    print (response.text)
else:
    accessToken = json.loads(response.text)['access_token']
    print(accessToken)

validate the token

 header = {
    "Authorization": "Bearer " + access_token
} 
userinfoUrl = 'https://localhost:8080/auth/realms/customer/protocol/openid-connect/userinfo'

response = req.get(userinfoUrl, headers=header)    
print(response.text)

Salesforce Rest API 查找数据

官方文档:

https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/using_resources_working_with_records.htm

如何生成access token? 其中client id 和 client secret可以在connected app里找到,username和password为Salesforce里一个user的用户名和 密码 + security token


import requests as req
import json
 
testTokenUrl = "https://test.salesforce.com/services/oauth2/token"
testReqJson = {
    "client_id": "xxx",
    "client_secret": "xxx",
    "username": "sandbox_dev",
    "password": "password+securityToken",
    "grant_type": "password"
}
 
header = {
    "Content-Type": "application/x-www-form-urlencoded",
}
 
response = req.post(testTokenUrl, data=testReqJson, headers=header)
if response.status_code > 200:
    print (response.text)
else:
    token = json.loads(response.text)
    accessToken = json.loads(response.text)['access_token']

比较通用的方法是通过sql语句查询所需的数据

def get_record_type_id(token):
    header = {
        "Authorization": "Bearer " + token['access_token']
    }
    queryURL = token['instance_url'] + "/services/data/v47.0/query/?q="
    queryParams = "Select+Id+FROM+RecordType+WHERE+Name='Root Branch'"
    response = req.get(queryURL+queryParams, headers=header)
    
    if response.status_code == 200:
        return json.loads(response.text)['records'][0]['Id']
    else:
        raise ImportException('Error: '+response.text)

Salesforce Rest API 更新数据

官方文档:

https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_update_fields.htm

如何生成access token?其中client id 和 client secret可以在connected app里找到,username和password为Salesforce里一个user的用户名和密码,

import requests as req
import json

testTokenUrl = "https://test.salesforce.com/services/oauth2/token"
testReqJson = {
    "client_id": "xxx",
    "client_secret": "xxx",
    "username": "sandbox_dev",
    "password": "password+securityToken",
    "grant_type": "password"
}

header = {
    "Content-Type": "application/x-www-form-urlencoded",
}

response = req.post(testTokenUrl, data=testReqJson, headers=header)
if response.status_code > 200:
    print (response.text)
else:
    token = json.loads(response.text)
    accessToken = json.loads(response.text)['access_token']
    print(accessToken)

更新记录 Update a Record

curl https://yourInstance.salesforce.com/services/data/v20.0/sobjects/Account/001D000000INjVe -H “Authorization: Bearer token” -H “Content-Type: application/json” -d @patchaccount.json -X PATCH

如果SF返回的http code为204,说明更新成功

def update_record_type_id(token, RecordTypeId, loanOfficeNameId):
    print('Set record type id...')
    header = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + token['access_token']
    }
    params = {
        "RecordTypeId": RecordTypeId
    }
    body=json.JSONEncoder().encode(params)
    patchURL = token['instance_url'] + "/services/data/v47.0/sobjects/loan__Office_Name__c/"+loanOfficeNameId
    response = req.patch(patchURL,data=body, headers=header)
    if response.status_code == 204:
        print('set_record_type_id Done.')
    else:
        raise ImportException('Error: '+response.text)

更新user相关的数据

def set_user_branch(branchName='aaa'):
    token = get_token()
    header = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + token['access_token']
    }
    params = {
        "loan__Current_Branch__c": branchName,
        "loan__Default_Branch__c": branchName
    }
    index = token['id'].rfind('/')
    body=json.JSONEncoder().encode(params)
    patchURL = token['instance_url'] + "/services/data/v{}/sobjects/User/{}".format(SF_API_VERSION, token['id'][index+1:])
    #如果知道user id,可以简化为
    # patchURL = token['instance_url'] + "/services/data/v47.0/sobjects/User/"+Id
    response = req.patch(patchURL,data=body, headers=header)
    if response.status_code == 204:
        print('Done.')
    else:
        raise ImportException('Error: '+response.text)