Comparison
SDK vs Raw Requests
See side-by-side how the PiSharp Python SDK compares to writing raw HTTP requests for common PI Web API tasks. The SDK eliminates boilerplate while preserving an escape hatch for raw access when you need it.
When to use which
| Factor | Raw requests | SDK |
|---|---|---|
| Setup time | Medium -- session, auth, SSL, headers, retry | Low -- one constructor call |
| Boilerplate per operation | High -- URL building, JSON parsing, error checks, pagination | Low -- methods return typed results |
| Flexibility | Full -- any endpoint, any parameter | High -- common tasks covered, pi.raw_request() for everything else |
| Error handling | Manual -- check status codes, parse error bodies, handle edge cases | Built in -- typed exceptions (PIAuthError, PINotFoundError) |
| Pagination | Manual -- follow links, manage state, detect truncation | Automatic -- transparent iteration with truncation warnings |
| Digital states | Manual -- detect JSON objects, extract Name field | Automatic -- converted to strings or filtered |
| Dependencies | requests only | pisharp-piwebapi (wraps requests) |
| Best for | One-off scripts, unusual endpoints, learning the API | Production integrations, team projects, repeated tasks |
Comparison: Connect and authenticate
Setting up a session with authentication, SSL, connection pooling, and retry logic.
Raw requests (12 lines)python
import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
session.auth = (os.environ["PI_USERNAME"], os.environ["PI_PASSWORD"])
session.verify = os.environ.get("PI_CA_BUNDLE", True)
session.headers.update({"Accept": "application/json"})
retries = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retries, pool_connections=10, pool_maxsize=10)
session.mount("https://", adapter)
BASE_URL = os.environ["PI_WEB_API_URL"]With SDK (5 lines)python
import os
from pisharp_piwebapi import PIWebAPI
pi = PIWebAPI(
base_url=os.environ["PI_WEB_API_URL"],
username=os.environ["PI_USERNAME"],
password=os.environ["PI_PASSWORD"],
ca_bundle=os.environ.get("PI_CA_BUNDLE"),
)
# Retry logic, connection pooling, and session management are automaticComparison: Find a PI point
Look up a PI point by path and retrieve its WebID.
Raw requests (5 lines)python
path = "\\\\MY-SERVER\\sinusoid"
resp = session.get(
f"{BASE_URL}/points",
params={"path": path, "selectedFields": "WebId;Name;Path;PointType"},
)
resp.raise_for_status()
data = resp.json()
web_id = data["WebId"]
name = data["Name"]With SDK (1 line)python
point = pi.points.get("sinusoid", server="MY-SERVER")
# point.web_id, point.name, point.path, point.point_type all availableComparison: Read current value
Get the latest snapshot value for a PI point, handling digital states correctly.
Raw requests (11 lines)python
resp = session.get(
f"{BASE_URL}/streams/{web_id}/value",
params={"selectedFields": "Value;Timestamp;Good;UnitsAbbreviation"},
)
resp.raise_for_status()
data = resp.json()
# Handle digital state values (they come back as JSON objects)
raw_value = data["Value"]
if isinstance(raw_value, dict):
value = raw_value.get("Name", str(raw_value)) # Digital state
else:
value = raw_value
timestamp = data["Timestamp"]
good = data.get("Good", True)With SDK (1 line)python
value = pi.points.get_value("sinusoid", server="MY-SERVER")
# Returns a typed result with .value, .timestamp, .good, .units
# Digital states are already converted to stringsComparison: Read recorded values to DataFrame
Pull 24 hours of recorded history into a pandas DataFrame with quality filtering and truncation detection.
Raw requests (20 lines)python
import pandas as pd
resp = session.get(
f"{BASE_URL}/streams/{web_id}/recorded",
params={
"startTime": "*-24h",
"endTime": "*",
"maxCount": 10000,
"selectedFields": "Items.Value;Items.Timestamp;Items.Good;Links",
},
)
resp.raise_for_status()
data = resp.json()
# Check for silent truncation
if "Next" in data.get("Links", {}):
print("WARNING: Data was truncated -- more pages available")
items = data["Items"]
# Filter out digital states and bad quality values
clean_items = []
for item in items:
if not item.get("Good", True):
continue
val = item["Value"]
if isinstance(val, dict):
continue # Skip digital states for numeric analysis
clean_items.append({"Timestamp": item["Timestamp"], "Value": val})
df = pd.DataFrame(clean_items)
df["Timestamp"] = pd.to_datetime(df["Timestamp"], utc=True)
df = df.set_index("Timestamp").sort_index()With SDK (5 lines)python
df = pi.points.get_recorded(
"sinusoid",
server="MY-SERVER",
start_time="*-24h",
end_time="*",
)
# Returns a clean pandas DataFrame indexed by timestamp
# Quality filtering, digital state handling, and truncation warnings automaticComparison: Batch read 100+ points
Read current values for many points in one operation, with automatic chunking and error handling.
Raw requests (20 lines)python
import time
results = {}
chunk_size = 100
for i in range(0, len(web_ids), chunk_size):
chunk = web_ids[i : i + chunk_size]
batch = {}
for j, wid in enumerate(chunk):
batch[f"req_{i + j}"] = {
"Method": "GET",
"Resource": f"{BASE_URL}/streams/{wid}/value"
f"?selectedFields=Value;Timestamp;Good",
}
resp = session.post(f"{BASE_URL}/batch", json=batch)
resp.raise_for_status()
for key, result in resp.json().items():
status = result.get("Status", 0)
if status == 200:
results[key] = result["Content"]
else:
print(f"Failed: {key} -> HTTP {status}")
if i + chunk_size < len(web_ids):
time.sleep(0.2) # Rate limitingWith SDK (1 line)python
values = pi.batch.read_current(web_ids)
# Automatic chunking, error handling, rate limiting, and typed resultsComparison: Write a value
Write a single timestamped value to a PI point.
Raw requests (7 lines)python
from datetime import datetime, timezone
payload = {
"Value": 72.5,
"Timestamp": datetime.now(timezone.utc).isoformat(),
}
resp = session.post(f"{BASE_URL}/streams/{web_id}/value", json=payload)
if not resp.ok:
error = resp.json().get("Errors", [resp.text])
raise RuntimeError(f"Write failed: HTTP {resp.status_code} - {error}")With SDK (1 line)python
pi.points.write_value("sinusoid", server="MY-SERVER", value=72.5)
# Raises PIWriteError with context on failureComparison: Error handling
Handle different failure modes with meaningful error messages.
Raw requests (15 lines)python
resp = session.get(f"{BASE_URL}/streams/{web_id}/recorded",
params={"startTime": "*-1h"})
if resp.status_code == 401:
raise RuntimeError("Authentication failed -- check credentials")
elif resp.status_code == 403:
raise RuntimeError("Permission denied -- check PI identity mapping")
elif resp.status_code == 404:
raise RuntimeError(f"Point not found -- WebID may be stale: {web_id}")
elif resp.status_code == 500:
errors = resp.json().get("Errors", [])
raise RuntimeError(f"Server error: {errors}")
elif resp.status_code == 503:
raise RuntimeError("Server overloaded -- reduce query size or add backoff")
elif resp.status_code != 200:
raise RuntimeError(f"Unexpected error: HTTP {resp.status_code}")
data = resp.json()With SDK (6 lines)python
from pisharp_piwebapi.errors import PIAuthError, PINotFoundError, PIServerError
try:
df = pi.points.get_recorded("sinusoid", server="MY-SERVER", start_time="*-1h")
except PIAuthError as e:
print(f"Auth failed: {e}") # Includes which auth method was attempted
except PINotFoundError as e:
print(f"Not found: {e}") # Includes the path that was looked up
except PIServerError as e:
print(f"Server error: {e}") # Includes the PI Web API error messagesLine count comparison
| Task | Raw requests | SDK | Reduction |
|---|---|---|---|
| Connect + auth + retry | 12 lines | 5 lines | 58% |
| Find a point | 5 lines | 1 line | 80% |
| Read current value (with digital state handling) | 11 lines | 1 line | 91% |
| Recorded to DataFrame (with quality + truncation) | 20 lines | 5 lines | 75% |
| Batch read 100+ points | 20 lines | 1 line | 95% |
| Write a value | 7 lines | 1 line | 86% |
| Error handling | 15 lines | 6 lines | 60% |
Total for a typical integration
A typical PI Web API integration (connect, find 10 points, read recorded values, write results) takes approximately 80 lines with raw requests vs 15 lines with the SDK -- an 81% reduction. The biggest wins come from batch operations, error handling, and DataFrame conversion, where the SDK handles edge cases you would otherwise need to discover and fix one production incident at a time.
Ready to try the SDK?
Visit the SDK page for installation instructions and the full API reference, or check the GitHub repository for source code and examples.