feat(aveva_pi): Add AVEVA PI Connector SDK connector

README.md

@@ -31,6 +31,7 @@ For SDK installation and setup, visit the main [Fivetran Connector SDK repositor
 - **[apache_hive/using_pyhive](https://github.com/fivetran/community_connectors/tree/main/apache_hive/using_pyhive)** - Sync data from Apache Hive using PyHive
 - **[apache_hive/using_sqlalchemy](https://github.com/fivetran/community_connectors/tree/main/apache_hive/using_sqlalchemy)** - Sync data from Apache Hive using SQLAlchemy with PyHive
 - **[arango_db](https://github.com/fivetran/community_connectors/tree/main/arango_db)** - Sync document and edge collections from ArangoDB multi-model database
+- **[aveva_pi](https://github.com/fivetran/community_connectors/tree/main/aveva_pi)** - Sync asset hierarchy and time-series data from AVEVA PI (formerly OSIsoft PI) via PI Web API
 - **[cassandra](https://github.com/fivetran/community_connectors/tree/main/cassandra)** - Connect and sync data from Cassandra database
 - **[clickhouse](https://github.com/fivetran/community_connectors/tree/main/clickhouse)** - Sync data from ClickHouse database
 - **[couchbase_capella](https://github.com/fivetran/community_connectors/tree/main/couchbase_capella)** - Sync data from Couchbase Capella

aveva_pi/README.md

@@ -0,0 +1,162 @@
+# AVEVA PI Connector SDK Connector Example
+
+## Connector overview
+
+This connector syncs data from AVEVA PI (formerly OSIsoft PI) to your Fivetran destination. It communicates with the PI system via the PI Web API REST interface — no proprietary ODBC drivers are required, so the connector runs in Fivetran's managed cloud environment without any additional installation.
+
+Key capabilities:
+- REST-based connectivity via PI Web API (HTTPS + Basic auth) — no ODBC driver installation required
+- Four fixed tables: `elements`, `attributes`, `event_frames`, and `recorded_values`
+- Full reimport for `elements` and `attributes` (PI AF asset hierarchy)
+- Cursor-based incremental sync for `event_frames` and `recorded_values` with adaptive time-window backoff
+- 2-hour late-arrival rollback on `recorded_values` to capture values written after their timestamps
+- MD5-based synthetic `_fivetran_id` primary key for `recorded_values`
+- Authentication-aware retry: 4xx responses surface immediately; 5xx / network errors retry up to 3 times
+- `recorded_values` sync is opt-in (set `sync_recorded_values = "true"`) because it can generate very large data volumes
+
+
+## Requirements
+
+- [Supported Python versions](https://github.com/fivetran/community_connectors/blob/main/README.md#requirements)
+- Operating system:
+  - Windows: 10 or later (64-bit only)
+  - macOS: 13 (Ventura) or later (Apple Silicon [arm64] or Intel [x86_64])
+  - Linux: Distributions such as Ubuntu 20.04 or later, Debian 10 or later, or Amazon Linux 2 or later (arm64 or x86_64)
+- PI Web API 2019 SP1 or later, reachable over HTTPS from the connector host
+- Basic authentication enabled on the PI Web API server
+- A PI user account with read access to the target AF database
+
+## Getting started
+
+Refer to the [Connector SDK Setup Guide](https://fivetran.com/docs/connectors/connector-sdk/setup-guide) to get started.
+
+Running `fivetran init --template aveva_pi` creates a new Connector SDK project pre-populated with this connector's source files. You can then update `configuration.json` with your PI Web API credentials and run `fivetran debug` to test locally against your own PI server.
+
+
+## Configuration file
+
+```json
+{
+  "base_url": "<PI_WEB_API_BASE_URL>",
+  "username": "<PI_USERNAME>",
+  "password": "<PI_PASSWORD>",
+  "database_name": "<PI_AF_DATABASE_NAME>",
+  "verify_ssl": "<TRUE_OR_FALSE_DEFAULT_TRUE>",
+  "start_date": "<START_DATE_ISO8601_EXAMPLE_2020_01_01T00_00_00Z>",
+  "sync_recorded_values": "<TRUE_OR_FALSE_DEFAULT_FALSE>"
+}
+```
+
+| Key | Required | Description |
+|---|---|---|
+| `base_url` | Yes | Base URL of the PI Web API instance (e.g. `https://piserver/piwebapi`) |
+| `username` | Yes | PI user account with read access to the target AF database |
+| `password` | Yes | Password for the PI user account |
+| `database_name` | No | PI AF database name to sync. Defaults to the first database found if omitted |
+| `verify_ssl` | No | Set to `"false"` to skip TLS certificate verification for self-signed certificates (default: `"true"`) |
+| `start_date` | No | ISO 8601 start date for the first incremental sync (default: Unix epoch). Example: `"2020-01-01T00:00:00Z"` |
+| `sync_recorded_values` | No | Set to `"true"` to also sync the `recorded_values` table. Disabled by default because it can generate very large data volumes on large PI deployments |
+
+> Note: When submitting connector code as a Community Connector, ensure `configuration.json` has placeholder values. When deploying, do not check this file into version control to protect credentials.
+
+
+## Authentication
+
+The connector uses HTTP Basic authentication. To set up credentials:
+
+1. Log in to your PI Web API server admin interface and confirm that Basic authentication is enabled under Security settings.
+2. Create or identify a PI user account with read access to the target AF database.
+3. Add the `username` and `password` for that account to `configuration.json`.
+4. If your PI Web API server uses a self-signed TLS certificate, set `verify_ssl` to `"false"` in `configuration.json`.
+
+
+## Pagination
+
+PI Web API responses include a `Links.Next` URL when there are more items. The connector follows this link chain automatically until all items are retrieved. Each page is processed and yielded immediately — no full result set is held in memory.
+
+For incremental tables (`event_frames` and `recorded_values`), data is fetched in time windows of up to 30 days. If a request fails with a transient error, the window is halved and the same slice is retried. This continues until either the request succeeds or the window drops below 1 hour, at which point the error is surfaced.
+
+Checkpointing occurs after each successful time window (incremental) or every 10,000 rows (full reimport), allowing the connector to resume from the last safe point after an interruption.
+
+
+## Data handling
+
+- Schema: Fixed four-table schema. Column types are auto-detected by the Fivetran Connector SDK from the data values upserted during sync.
+- Timestamps: PI Web API returns ISO 8601 timestamps. The connector parses them to UTC-aware `datetime` objects before yielding rows to Fivetran.
+- PI digital states: When a PI recorded value is a system digital state (a JSON object like `{"Name": "Shutdown", "Value": 248}`), only the `Name` string is stored in the `value` column.
+- Hash IDs: The `recorded_values` table has no natural primary key. A `_fivetran_id` column is generated as the MD5 hex digest of `attribute_web_id|timestamp`.
+- Category names: The `category_names` column stores a JSON-serialized array of category name strings (e.g. `["Production", "Critical"]`).
+
+
+## Error handling
+
+- HTTP 4xx responses raise a `ValueError` immediately (no retry), except for 408 (Request Timeout) and 429 (Too Many Requests) which are treated as transient and retried. Refer to `api_get()` in `client.py`.
+- HTTP 5xx and network errors retry up to 3 times with a warning logged per attempt. Refer to `api_get()` in `client.py`.
+- Incremental query failures trigger adaptive window halving rather than a hard failure. If the window cannot be halved further (below 1 hour), a `RuntimeError` is raised. Refer to `sync_event_frames()` and `sync_recorded_values()` in `sync.py`.
+- Individual `recorded_values` attribute streams that return 4xx errors are skipped with a warning (e.g. deleted PI Points). Refer to `sync_recorded_values()` in `sync.py`.
+
+
+## Tables created
+
+### elements
+
+Full reimport. Represents PI AF elements (the asset hierarchy).
+
+| Column | Type | Primary key |
+|---|---|---|
+| `web_id` | STRING | Yes |
+| `name` | STRING | |
+| `description` | STRING | |
+| `path` | STRING | |
+| `template_name` | STRING | |
+| `category_names` | STRING | JSON array, e.g. `["Production"]` |
+
+### attributes
+
+Full reimport. Represents PI AF element attributes, including PI Point data reference metadata.
+
+| Column | Type | Primary key |
+|---|---|---|
+| `web_id` | STRING | Yes |
+| `element_web_id` | STRING | |
+| `name` | STRING | |
+| `description` | STRING | |
+| `path` | STRING | |
+| `type` | STRING | |
+| `type_qualifier` | STRING | |
+| `data_reference` | STRING | e.g. `"PI Point"` |
+| `data_reference_path` | STRING | PI Point tag path |
+| `category_names` | STRING | JSON array |
+
+### event_frames
+
+Incremental by `start_time`. Represents PI AF event frames (time-bounded events).
+
+| Column | Type | Primary key |
+|---|---|---|
+| `web_id` | STRING | Yes |
+| `name` | STRING | |
+| `description` | STRING | |
+| `start_time` | UTC_DATETIME | |
+| `end_time` | UTC_DATETIME | |
+| `template_name` | STRING | |
+| `category_names` | STRING | JSON array |
+| `database_web_id` | STRING | |
+
+### recorded_values
+
+Incremental by `timestamp`. Opt-in via `sync_recorded_values = "true"`. Represents PI archive (time-series) data for PI Point attributes.
+
+| Column | Type | Primary key |
+|---|---|---|
+| `_fivetran_id` | STRING | Yes — MD5 of `attribute_web_id\|timestamp` |
+| `attribute_web_id` | STRING | |
+| `timestamp` | UTC_DATETIME | |
+| `value` | STRING | |
+| `quality` | STRING | `"good"` or `"questionable"` |
+| `good` | BOOLEAN | |
+
+
+## Additional considerations
+
+The examples provided are intended to help you effectively use Fivetran's Connector SDK. While we've tested the code, Fivetran cannot be held responsible for any unexpected or negative consequences that may arise from using these examples. For inquiries, please reach out to our Support team.

aveva_pi/client.py

@@ -0,0 +1,153 @@
+# For making HTTP requests to the PI Web API
+import time  # For sleep-based backoff between retry attempts
+import requests
+
+# For HTTP Basic authentication
+from requests.auth import HTTPBasicAuth
+
+# For enabling logs in the connector
+from fivetran_connector_sdk import Logging as log
+
+# Maximum retry attempts for transient server/network errors before raising
+__MAX_RETRIES = 3
+
+
+def build_session(configuration: dict) -> requests.Session:
+    """
+    Create an authenticated requests.Session for PI Web API calls.
+
+    Args:
+        configuration: a dictionary that holds the configuration settings for the connector.
+    Returns:
+        A configured requests.Session with Basic auth and JSON accept headers.
+    """
+    session = requests.Session()
+    session.auth = HTTPBasicAuth(configuration["username"], configuration["password"])
+    session.headers.update(
+        {
+            "Accept": "application/json",
+            "X-Requested-With": "XMLHttpRequest",
+        }
+    )
+    # Allow users to disable TLS verification for self-signed PI Web API certificates.
+    # Only enable verification when the value is explicitly "true".
+    session.verify = str(configuration.get("verify_ssl", "true")).lower() == "true"
+    return session
+
+
+def base_url(configuration: dict) -> str:
+    """
+    Return the PI Web API base URL with any trailing slash removed.
+
+    Args:
+        configuration: a dictionary that holds the configuration settings for the connector.
+    Returns:
+        The base URL string.
+    """
+    return configuration["base_url"].rstrip("/")
+
+
+def api_get(session: requests.Session, url: str, params: dict = None) -> dict:
+    """
+    GET a PI Web API endpoint and return the parsed JSON body.
+
+    Raises ValueError immediately on 4xx responses (auth failures, not-found) —
+    these are not worth retrying. Retries up to __MAX_RETRIES times on 5xx or
+    network/connection errors using exponential backoff (honoring Retry-After when present).
+
+    Args:
+        session: an authenticated requests.Session.
+        url: the full URL to GET.
+        params: optional query parameters dict.
+    Returns:
+        Parsed JSON response body as a dict.
+    Raises:
+        ValueError: on 4xx HTTP responses.
+        requests.exceptions.ConnectionError: after __MAX_RETRIES consecutive transient failures.
+    """
+    last_exc: Exception = RuntimeError("No request attempted")
+    for attempt in range(1, __MAX_RETRIES + 1):
+        try:
+            resp = session.get(url, params=params, timeout=30)
+            if resp.status_code in (401, 403):
+                raise ValueError(
+                    f"Authentication error ({resp.status_code}) for {url}: {resp.text[:200]}"
+                )
+            # 408 (Request Timeout) and 429 (Too Many Requests) are transient — retry them
+            if resp.status_code in (408, 429):
+                raise requests.exceptions.HTTPError(
+                    f"Retryable HTTP {resp.status_code} for {url}", response=resp
+                )
+            if 400 <= resp.status_code < 500:
+                raise ValueError(f"Client error ({resp.status_code}) for {url}: {resp.text[:200]}")
+            resp.raise_for_status()
+            return resp.json()
+        except ValueError:
+            # Auth / client errors — surface immediately, no retry
+            raise
+        except requests.exceptions.RequestException as exc:
+            last_exc = exc
+            log.warning(f"Request attempt {attempt}/{__MAX_RETRIES} failed for {url}: {exc}")
+            if attempt < __MAX_RETRIES:
+                time.sleep(min(60, 2 ** (attempt - 1)))
+
+    raise requests.exceptions.ConnectionError(
+        f"Could not reach PI Web API after {__MAX_RETRIES} attempts. URL: {url}"
+    ) from last_exc
+
+
+def paginate(session: requests.Session, url: str, params: dict = None):
+    """
+    Yield every item from a paginated PI Web API response.
+
+    PI Web API paginates via a 'Links.Next' URL embedded in each response body.
+    Parameters are only sent with the first request; subsequent pages use the
+    full Next URL returned by the API.
+
+    Args:
+        session: an authenticated requests.Session.
+        url: the initial URL to GET.
+        params: optional query parameters for the first request.
+    """
+    next_url = url
+    next_params = params
+    while next_url:
+        body = api_get(session, next_url, next_params)
+        for item in body.get("Items", []):
+            yield item
+        next_url = body.get("Links", {}).get("Next")
+        next_params = None  # Parameters are already encoded in the Next URL
+
+
+def get_database_web_id(session: requests.Session, base: str, database_name: str) -> str:
+    """
+    Find and return the WebId of the target AF database.
+
+    Searches all asset servers visible to this PI Web API instance. If
+    database_name is provided, returns the first database with that exact name.
+    If database_name is None, returns the first database found on any server.
+
+    Args:
+        session: an authenticated requests.Session.
+        base: the PI Web API base URL.
+        database_name: target AF database name, or None to use the first found.
+    Returns:
+        The WebId string of the matching database.
+    Raises:
+        ValueError: if no matching database can be found.
+    """
+    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.")

aveva_pi/configuration.json

@@ -0,0 +1,9 @@
+{
+  "base_url": "<PI_WEB_API_BASE_URL>",
+  "username": "<PI_USERNAME>",
+  "password": "<PI_PASSWORD>",
+  "database_name": "<PI_AF_DATABASE_NAME>",
+  "verify_ssl": "<TRUE_OR_FALSE_DEFAULT_TRUE>",
+  "start_date": "<START_DATE_ISO8601_EXAMPLE_2020_01_01T00_00_00Z>",
+  "sync_recorded_values": "<TRUE_OR_FALSE_DEFAULT_FALSE>"
+}