Start Here: Your First PI Web API Request

import requests
from requests.auth import HTTPBasicAuth

BASE_URL = "https://your-server/piwebapi"

response = requests.get(
    BASE_URL,
    auth=HTTPBasicAuth("DOMAIN\\username", "password"),
    verify=False,  # See the Authentication guide for proper cert handling
)

print(f"Status:  {response.status_code}")

if response.status_code == 200:
    data = response.json()
    print(f"Product: {data['ProductTitle']}")
    print(f"Version: {data['ProductVersion']}")
    print(f"Server:  {data['MachineName']}")
    print(f"\nAvailable resources:")
    for name, url in data["Links"].items():
        if name != "Self":
            print(f"  {name}: {url}")
else:
    print(f"Error:   {response.text[:200]}")

Status:  200
Product: PI Web API
Version: 2021 SP3
Server:  YOURSERVER

Available resources:
  AssetServers: https://your-server/piwebapi/assetservers
  DataServers: https://your-server/piwebapi/dataservers
  Search: https://your-server/piwebapi/search
  System: https://your-server/piwebapi/system
  ...

# List all PI Data Archive servers this PI Web API can reach
response = requests.get(
    f"{BASE_URL}/dataservers",
    auth=HTTPBasicAuth("DOMAIN\\username", "password"),
    verify=False,
)

servers = response.json()["Items"]
for server in servers:
    print(f"Server: {server['Name']}")
    print(f"  WebId:   {server['WebId']}")
    print(f"  Path:    {server['Path']}")
    print(f"  IsConnected: {server['IsConnected']}")
    print()

# Look up a PI point by its full path
# Replace YOUR-SERVER with your PI Data Archive server name
response = requests.get(
    f"{BASE_URL}/points",
    params={"path": "\\\\YOUR-SERVER\\sinusoid"},
    auth=HTTPBasicAuth("DOMAIN\\username", "password"),
    verify=False,
)

if response.status_code == 200:
    point = response.json()
    print(f"Name:      {point['Name']}")
    print(f"WebId:     {point['WebId']}")
    print(f"PointType: {point['PointType']}")
    print(f"Descriptor: {point.get('Descriptor', 'N/A')}")
    print(f"EngUnits:  {point.get('EngineeringUnits', 'N/A')}")

    # Save the WebId for the next step
    POINT_WEB_ID = point["WebId"]
elif response.status_code == 404:
    print("Point not found. Check the server name and point name.")
    print("Tip: Server name is case-sensitive on some configurations.")
else:
    print(f"Error {response.status_code}: {response.text[:200]}")

# Search for points by name (requires Indexed Search to be enabled)
response = requests.get(
    f"{BASE_URL}/search/query",
    params={"q": "name:sinusoid", "count": 5},
    auth=HTTPBasicAuth("DOMAIN\\username", "password"),
    verify=False,
)

if response.status_code == 200:
    results = response.json()
    items = results.get("Items", [])
    if items:
        for item in items:
            print(f"{item['Name']} ({item['ItemType']}) - WebId: {item['WebId']}")
    else:
        print("No results. The Indexed Search crawler may not be configured.")
        print("Use the path-based lookup above instead.")
elif response.status_code == 502:
    print("Search service unavailable. Indexed Search may not be enabled.")
    print("Use the path-based lookup above instead.")

# Read the current snapshot value
response = requests.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/value",
    auth=HTTPBasicAuth("DOMAIN\\username", "password"),
    verify=False,
)

data = response.json()
print(f"Value:     {data['Value']}")
print(f"Timestamp: {data['Timestamp']}")
print(f"Good:      {data['Good']}")
print(f"Units:     {data.get('UnitsAbbreviation', 'N/A')}")

# The full response contains more fields:
# - Substituted: whether the value was manually overridden
# - Annotated: whether the value has a note attached
# - Good: whether the data quality is acceptable

{
  "Timestamp": "2026-03-15T14:30:00Z",
  "Value": 87.345,
  "UnitsAbbreviation": "",
  "Good": true,
  "Questionable": false,
  "Substituted": false,
  "Annotated": false
}

# Read recorded values from the last hour
response = requests.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/recorded",
    params={
        "startTime": "*-1h",   # 1 hour ago
        "endTime": "*",        # Now
        "maxCount": 100,       # Limit results
    },
    auth=HTTPBasicAuth("DOMAIN\\username", "password"),
    verify=False,
)

data = response.json()
items = data.get("Items", [])

print(f"Retrieved {len(items)} recorded values:\n")
for item in items[:5]:  # Show first 5
    quality = "Good" if item["Good"] else "BAD"
    print(f"  {item['Timestamp']}: {item['Value']:.3f} [{quality}]")

if len(items) > 5:
    print(f"  ... and {len(items) - 5} more")

# Important: if len(items) == maxCount, there may be more data
# that was truncated. See the Pagination guide for how to handle this.
if len(items) == 100:
    print("\nWarning: results may be truncated. Increase maxCount or paginate.")

"""PI Web API Quick Start -- your first requests in Python.

Replace the configuration values below with your actual server details.
Run: python pi_web_api_quickstart.py
"""

import requests
from requests.auth import HTTPBasicAuth
import urllib3


# Suppress SSL warnings for this quick test
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# ── Configuration (replace these) ──────────────────────────────
BASE_URL   = "https://your-server/piwebapi"
USERNAME   = "DOMAIN\\username"
PASSWORD   = "password"
PI_SERVER  = "YOUR-PI-SERVER"   # PI Data Archive server name
POINT_NAME = "sinusoid"         # PI point to read
# ───────────────────────────────────────────────────────────────

session = requests.Session()
session.auth = HTTPBasicAuth(USERNAME, PASSWORD)
session.verify = False
session.headers.update({"Accept": "application/json"})

# Step 1: Verify connectivity
print("1. Connecting to PI Web API...")
resp = session.get(BASE_URL)
if resp.status_code != 200:
    print(f"   FAILED: {resp.status_code} - {resp.text[:200]}")
    exit(1)

info = resp.json()
print(f"   Connected to {info['ProductTitle']} {info['ProductVersion']}")
print(f"   Server: {info['MachineName']}")

# Step 2: Find the PI point
print(f"\n2. Looking up point '{POINT_NAME}' on {PI_SERVER}...")
resp = session.get(
    f"{BASE_URL}/points",
    params={"path": f"\\\\{PI_SERVER}\\{POINT_NAME}"},
)
if resp.status_code != 200:
    print(f"   FAILED: {resp.status_code} - {resp.text[:200]}")
    exit(1)

point = resp.json()
web_id = point["WebId"]
print(f"   Found: {point['Name']} (Type: {point['PointType']})")
print(f"   WebId: {web_id}")

# Step 3: Read the current value
print("\n3. Reading current value...")
resp = session.get(f"{BASE_URL}/streams/{web_id}/value")
data = resp.json()
quality = "Good" if data["Good"] else "BAD"
print(f"   Value: {data['Value']} [{quality}]")
print(f"   Timestamp: {data['Timestamp']}")

# Step 4: Read the last hour of history
print("\n4. Reading last hour of recorded values...")
resp = session.get(
    f"{BASE_URL}/streams/{web_id}/recorded",
    params={"startTime": "*-1h", "endTime": "*", "maxCount": 100},
)
items = resp.json().get("Items", [])
print(f"   Retrieved {len(items)} recorded values")

if items:
    values = [item["Value"] for item in items if item["Good"]]
    if values:
        print(f"   Min: {min(values):.3f}")
        print(f"   Max: {max(values):.3f}")
        print(f"   Avg: {sum(values)/len(values):.3f}")

print("\nDone! You have successfully connected to PI Web API.")