feat(aveva_pi): Add AVEVA PI Connector SDK connector#28
feat(aveva_pi): Add AVEVA PI Connector SDK connector#28fivetran-aishwaryardongal wants to merge 28 commits into
Conversation
🧹 Python Code Quality Check✅ No issues found in Python Files. This comment is auto-updated with every commit. |
There was a problem hiding this comment.
Pull request overview
Adds a new aveva_pi Connector SDK example that syncs AVEVA PI AF/Archive data via PI Web API (HTTPS/REST) instead of requiring ODBC drivers, aligning the example with Fivetran’s managed cloud environment constraints.
Changes:
- Introduces a PI Web API–based connector implementation with full-reimport + cursor-based incremental sync strategies.
- Adds example configuration + documentation describing setup, pagination, and table outputs.
- Adds connector folder scaffolding (
requirements.txt,.gitignore) consistent with a standalone example.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 20 comments.
Show a summary per file
| File | Description |
|---|---|
| aveva_pi/connector.py | Implements PI Web API session/auth, pagination, incremental cursors, and emits the four target tables. |
| aveva_pi/README.md | Documents connector behavior, configuration, pagination, and tables. |
| aveva_pi/configuration.json | Provides example configuration for PI Web API connectivity. |
| aveva_pi/requirements.txt | Notes that no additional dependencies are required. |
| aveva_pi/.gitignore | Adds local-environment ignores for the example folder. |
There was a problem hiding this comment.
You should not push this to github when adding it as example
There was a problem hiding this comment.
Acknowledged — the committed configuration.json uses angle-bracket placeholders only (e.g. <PI_WEB_API_BASE_URL>). No real credentials are present in the repository.
fivetran-JenasVimal
added
ai-assisted/yes
risk/low
and removed
risk/unknown
ai-assisted/unknown
labels
| except ValueError as exc: | ||
| # Surface 401 immediately — session-level auth failure affects all streams | ||
| if "Authentication error (401)" in str(exc): | ||
| raise | ||
| # 404 = PI Point deleted; 403 = no access to this stream — skip | ||
| log.warning(f" Skipping stream {attr_web_id}: {exc}") |
There was a problem hiding this comment.
Fixed. The per-stream handler now only skips on authentication error 403 (no access) and client error 404 (PI Point deleted). All other client errors — including 400 and 422 — are re-raised to surface real issues rather than silently advancing the cursor.
… case - Fix flake8 W503/black conflict by restructuring multi-line boolean expressions in sync.py - Tighten /elementattributes fallback to only trigger on 404/405 errors - Narrow per-stream skip to 403/404; surface all other errors including 401 - Fix start_date placeholder in configuration.json and README - Use sentence case for table headings in README per doc standards Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
| cursors = state.setdefault("cursors", {}) | ||
| start = parse_pi_timestamp(cursors.get("event_frames", start_date)) or datetime.fromtimestamp( | ||
| 0, tz=timezone.utc | ||
| ) |
There was a problem hiding this comment.
Fixed. The cursor handling now mirrors the sync_recorded_values approach: the stored cursor is parsed separately; if it is present but unparseable, a warning is logged and the fallback chain (start_date → epoch) is used instead of silently jumping to epoch.
| log.info("Syncing attributes (full reimport)") | ||
| count = 0 | ||
| pi_point_web_ids = [] | ||
|
|
There was a problem hiding this comment.
Acknowledged. The list is only materialized when sync_recorded_values = "true", which is opt-in and disabled by default. The README warns that enabling it can generate very large data volumes on large PI deployments. Streaming PI Point WebIds across the two sync functions would require a significant architectural change beyond the scope of this example connector, but the opt-in guard and the README warning together limit the blast radius for typical deployments.
| "password": "<PI_PASSWORD>", | ||
| "database_name": "<PI_AF_DATABASE_NAME>", | ||
| "verify_ssl": "<TRUE_OR_FALSE_DEFAULT_TRUE>", | ||
| "start_date": "<2020-01-01T00:00:00Z>", |
There was a problem hiding this comment.
Updated to <START_DATE_ISO8601_UTC_OPTIONAL> — descriptive, signals the expected format, and makes it clear this is user-supplied and optional.
| "password": "<PI_PASSWORD>", | ||
| "database_name": "<PI_AF_DATABASE_NAME>", | ||
| "verify_ssl": "<TRUE_OR_FALSE_DEFAULT_TRUE>", | ||
| "start_date": "<2020-01-01T00:00:00Z>", |
There was a problem hiding this comment.
Updated to <START_DATE_ISO8601_UTC_OPTIONAL> in the README config block as well.
| except ValueError as exc: | ||
| exc_str = str(exc) | ||
| # Surface auth failures immediately. | ||
| if "Authentication error" in exc_str: | ||
| raise |
There was a problem hiding this comment.
Fixed. Introduced PiApiError(status_code, message) in client.py. api_get() now raises PiApiError for all non-retryable 4xx responses, and all handlers in sync.py now branch on .status_code (e.g., exc.status_code == 401, exc.status_code not in (403, 404)) instead of parsing the error message string.
…, update placeholders - Add PiApiError(status_code, message) to client.py; api_get() now raises it for 4xx responses instead of ValueError, enabling callers to branch on status_code rather than parsing error message strings - Update all except-ValueError handlers in sync.py to except PiApiError and check .status_code directly (eliminates brittle substring matching) - Fix event_frames cursor fallback: malformed cursor now logs a warning and falls back to start_date (then epoch), matching the sync_recorded_values approach - Change start_date placeholder to <START_DATE_ISO8601_UTC_OPTIONAL> in both configuration.json and README to clearly signal it is user-supplied and optional Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
| if collect_pi_points and item.get("DataReferencePlugIn") == "PI Point": | ||
| pi_point_web_ids.append(item["WebId"]) |
There was a problem hiding this comment.
Fixed. Now uses .get("WebId", "") and only appends non-empty WebIds, matching the defensive style used elsewhere in the file.
| elem_web_id = elem_item.get("WebId", "") | ||
| try: | ||
| for attr_item in paginate( | ||
| session, | ||
| f"{base}/elements/{elem_web_id}/attributes", | ||
| params={"maxCount": __MAX_COUNT}, | ||
| ): |
There was a problem hiding this comment.
Fixed. Elements with a missing or empty WebId are now skipped with a warning before the /elements/{elem_web_id}/attributes URL is constructed.
… fallback fetch
- Use .get("WebId", "") and skip empty WebIds when accumulating PI Point attribute
WebIds to avoid KeyError when the API omits the field
- Skip elements with a missing WebId in the per-element fallback fetch to avoid
building an invalid URL and failing the whole sync
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
| servers = api_get(session, f"{base}/assetservers").get("Items", []) | ||
| if not servers: | ||
| raise ValueError("No PI Asset Servers found via PI Web API. Check base_url.") | ||
|
|
||
| for server in servers: | ||
| databases = api_get(session, f"{base}/assetservers/{server['WebId']}/assetdatabases").get( | ||
| "Items", [] | ||
| ) | ||
| for db in databases: | ||
| if database_name is None or db.get("Name") == database_name: | ||
| log.info(f"Connected to database '{db['Name']}' on server '{server.get('Name')}'") | ||
| return db["WebId"] | ||
|
|
||
| target = f"'{database_name}'" if database_name else "any database" | ||
| raise ValueError(f"Could not find {target} on any PI Asset Server. Check database_name.") |
There was a problem hiding this comment.
Fixed. get_database_web_id() now uses paginate() for both /assetservers and /assetdatabases, so all pages are walked. Server items with a missing WebId are skipped safely before the database URL is constructed.
| f"Invalid sync_recorded_values value '{sync_rv}'. Expected 'true' or 'false'." | ||
| ) | ||
|
|
||
| # Validate verify_ssl flag if provided — an unrecognised value silently disables TLS |
There was a problem hiding this comment.
Fixed. The comment now reads: build_session() keeps TLS enabled for any value other than "false", which accurately reflects the implementation.
…ading TLS comment - get_database_web_id() now uses paginate() for both /assetservers and /assetdatabases so all pages are walked on large deployments, and server items with a missing WebId are skipped safely - Correct the misleading verify_ssl comment in validate_configuration(): build_session() keeps TLS enabled for any value other than "false" Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
| for db in paginate(session, f"{base}/assetservers/{server_web_id}/assetdatabases"): | ||
| if database_name is None or db.get("Name") == database_name: | ||
| log.info(f"Connected to database '{db['Name']}' on server '{server.get('Name')}'") | ||
| return db.get("WebId", "") |
There was a problem hiding this comment.
Fixed. Database entries with a missing WebId are now skipped before returning, and db.get("Name", "") is used in both the match condition and the log statement, eliminating the KeyError risk.
…ame in discovery
- Skip database entries that have no WebId to avoid returning an empty string
and building invalid request URLs downstream
- Use db.get("Name", "") in both the match check and the log statement to
avoid KeyError if PI Web API omits the Name field
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
| table="recorded_values", | ||
| data=extract_recorded_value(item, attr_web_id), | ||
| ) | ||
| count += 1 |
There was a problem hiding this comment.
Fixed. Added a periodic op.checkpoint() every __CHECKPOINT_INTERVAL rows inside fetch_window, consistent with the approach in sync_elements and sync_attributes. This signals liveness for large windows and reduces rework on interruption.
…alueError in README - Add periodic op.checkpoint() every __CHECKPOINT_INTERVAL rows inside the fetch_window closure of sync_recorded_values() to signal liveness and reduce rework on large windows, consistent with sync_elements and sync_attributes - Update README error handling section: api_get() raises PiApiError (not ValueError) for non-retryable 4xx responses Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
| # Allow users to disable TLS verification for self-signed PI Web API certificates. | ||
| # Default to True (verification enabled). Only disable when the user explicitly sets | ||
| # verify_ssl to "false"; any other value (including template placeholders) keeps TLS on. | ||
| _verify_ssl = str(configuration.get("verify_ssl", "true")) | ||
| session.verify = False if _verify_ssl.lower() == "false" else True | ||
| return session |
There was a problem hiding this comment.
Fixed. build_session() now logs a log.warning() whenever TLS verification is disabled, making the security downgrade immediately visible in connector logs.
| log.info("Syncing attributes (full reimport)") | ||
| count = 0 | ||
| pi_point_web_ids = [] | ||
|
|
||
| def _process(item, element_web_id): | ||
| """ | ||
| Upsert one attribute record and optionally collect PI Point WebIds. | ||
|
|
||
| Args: | ||
| item: raw attribute dict from PI Web API. | ||
| element_web_id: WebId of the parent element, used as a foreign key. | ||
| """ | ||
| nonlocal count | ||
| # The 'upsert' operation is used to insert or update data in the destination table. | ||
| # The first argument is the name of the destination table. | ||
| # The second argument is a dictionary containing the record to be upserted. | ||
| op.upsert(table="attributes", data=extract_attribute(item, element_web_id)) | ||
| count += 1 | ||
| if collect_pi_points and item.get("DataReferencePlugIn") == "PI Point": | ||
| web_id = item.get("WebId", "") | ||
| if web_id: | ||
| pi_point_web_ids.append(web_id) | ||
| if count % __CHECKPOINT_INTERVAL == 0: |
There was a problem hiding this comment.
This is the same concern raised earlier (and already replied to). The list is only materialized when sync_recorded_values = "true", which is opt-in and disabled by default, and the README warns about large data volumes on large deployments. A full streaming refactor is out of scope for this example connector.
Makes the security downgrade explicit in connector logs when verify_ssl=false so accidental insecure configuration is immediately visible. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
| Args: | ||
| configuration: a dictionary that holds the configuration settings for the connector. | ||
| """ | ||
| validate_configuration(configuration) |
Summary
pyodbcdependency and the requirement to install AVEVA's proprietary ODBC driver — the connector now works in Fivetran's managed cloud environment with no additional installationWhat this connector does
elements(full reimport): PI AF elements — the asset hierarchy. UsesGET /assetdatabases/{webId}/elements?searchFullHierarchy=true.attributes(full reimport): PI AF element attributes including PI Point data reference metadata. Uses/elementattributes(PI Web API 2019+) with per-element fallback for older versions. Returns PI Point attribute WebIds for use by the recorded_values sync.event_frames(cursor-based incremental bystart_time): PI AF event frames. UsesGET /assetdatabases/{webId}/eventframes?startTime=...&endTime=.... Adaptive time-window backoff: starts at 30-day windows, halves on request failures, minimum 1-hour window.recorded_values(cursor-based incremental bytimestamp, opt-in): PI archive data for PI Point attributes. UsesGET /streams/{attrWebId}/recorded?startTime=...&endTime=...per attribute. Applies a 2-hour late-arrival rollback on subsequent syncs. Disabled by default (sync_recorded_values = "false") because it can generate very large data volumes.Design decisions
requests(no extra deps)requirements.txtis emptyschema()sync_recorded_valuesopt-inTesting
Syntax and import validation:
Local debug run (requires a reachable PI Web API instance):
# Update configuration.json with real PI Web API details fivetran debugValidated:
python -c "import connector"— no import errors ✓schema()returns all 4 tables with correct PKs and column types ✓requests✓Test plan
configuration.jsonwith a real PI Web API endpointfivetran debugand verify elements + attributes are upsertedfivetran debuga second time and verify incremental cursors advance for event_framessync_recorded_values = "true"and verify recorded_values are syncedScreenshots and testing
fivetran deploy --api-key $FIVETRAN_API_KEY --destination abdul_test_live_2 --connection ad_test3
