AnsweredAssumed Answered

Help with Python and PI Web API for a batch call to update PI Tag attributes

Question asked by jagdish.konathala Champion on Jul 6, 2020
Latest reply on Jul 7, 2020 by jagdish.konathala

I am working on automating this workflow rather than append in Excel. I have to set point and data security for some set of users or provide access to Analysis service. Got a great head-start from OSIsoft's Github. Adapted code for AF attributes to that of the PI Server points. Fetch the WebID and update attribute in the same call. Rather simple test prior to making it lot more interesting. I am not interested in C# & AF SDK since I am more comfortable in Python. 


"""This script creates and executes a batch call using the PI Web API
   This python script requires some pre-requisites:
   1.  A back-end server with PI WEB API with CORS enabled.

import json
import requests

from requests_kerberos import HTTPKerberosAuth


def call_headers(include_content_type):
    """ Create the header and return a string.
        include_content_type is a flag that determines whether or not the
        content-type header is included

    if include_content_type is True:
        header = {
            'content-type': 'application/json',
            'X-Requested-With': 'XmlHttpRequest'
        header = {
            'X-Requested-With': 'XmlHttpRequest'

    return header

def call_security_method():
    """ Create API call security method

    from requests_negotiate_sspi import HttpNegotiateAuth

    security_auth = HttpNegotiateAuth()
    return security_auth

def do_batch_call(piwebapiurl, pi_server):
    """ Create and execute a PI Web API batch call
        piwebapiurl: the URL of the PI Web API
        assetServer:  the AF server name
        user_name:  User's credentials name
        user_password:  User's credentials password
        piwebapi_security_method:  Security method:  basic or kerberos


    #  create security method - basic or kerberos
    security_method = call_security_method()

    point_name = 'sinusoid'

    #  Get the sample tag
    request_url = '{}/points?path=\\\\{}\\{}'.format(piwebapiurl, pi_server, point_name)
    response = requests.get(request_url, auth=security_method, verify='tls-ca-bundle.pem')

    #  Only continue if the first request was successful
    if response.status_code == 200:
        #  Deserialize the JSON Response
        data = json.loads(response.text)

        #  Create the header
        header = call_headers(False)

        attributeValue = 'piadmins: A(r,w) | Users: A(r)'

        #  Create the data for this call
        batch_request = {
            '1': {
                'Method': 'GET',
                'Resource': request_url,
                'Content': '{}'
            '2': {
                'Method': 'PUT',
                'Resource': piwebapiurl + '/points/{0}/attributes/datasecurity',
                'Content': '{\'Value\':' + attributeValue + '}',
                'Parameters': ['$.1.Content.WebId'],
                'ParentIds': ['1']

        #  Now that we have the attribute, we need to read the stream value
        response = + '/batch', auth=security_method, verify='tls-ca-bundle.pem',
                                 json=batch_request, headers=header)

        if response.status_code == 204:
            print('Batch Status: ' + str(response.status_code))

            #  Deserialize the JSON Response
            data = json.loads(response.text)

            #  Print the results for each call and format it so it is legible JSON
            print('1: Get the sample tag')
            print(json.dumps(data['1'], indent=4, sort_keys=True))

            print('2: Write pt attribute value to the sample tag')
            print(json.dumps(data['2'], indent=4, sort_keys=True))

            print(response.status_code, response.reason, response.text)
        print(response.status_code, response.reason, response.text)

    return response.status_code

def main():
    """ Main method.  Receive user input and call the do_batch_call method """

    do_batch_call(piwebapi_url, pi_server_name)

if __name__ == '__main__':


This is the output. The PUT change is not going through. It is fetching the WebID


207 Multi-Status {"1":{"Status":200,"Headers":{"Content-Type":"application/json; charset=utf-8"},"Content":{"WebId":"F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ","Id":152298,"Name":"SINUSOID","Path":"\\\\piserver\\SINUSOID","Descriptor":"12 Hour Sine Wave","PointClass":"classic","PointType":"Float32","DigitalSetName":"","EngineeringUnits":"","Span":100.0,"Zero":0.0,"Step":false,"Future":false,"DisplayDigits":-5,"Links":{"Self":"https://piwebapi/points/F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ","DataServer":"https:///piwebapi/dataservers/F1DSWc3DOup-50WOZAe8GD3RHASldUQ1ZQRU5UUEk","Attributes":"https://pivision/piwebapi/points/F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ/attributes","InterpolatedData":"https://piwebapi/streams/F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ/interpolated","RecordedData":"https://piwebapi/streams/F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ/recorded","PlotData":"https://piwebapi/streams/F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ/plot","SummaryData":"https://piwebapi/streams/F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ/summary","Value":"https://piwebapi/streams/F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ/value","EndValue":"https://piwebapi/streams/F1DPWc3DOup-50WOZAe8GD3RHA6lICAASldUQ1ZQRU5UUElcU0lOVVNPSUQ/end"}}},"2":{"Status":400,"Headers":{"Content-Type":"application/json; charset=utf-8"},"Content":{"Errors":["An exception has occurred. Please contact your PI Web API administrator for help in enabling debug mode."]}}}