Veeva Vault CDMS API Integration with Python

A first Veeva Vault CDMS integration usually breaks in one of three places: the version in the URL path drifts and a response field silently renames, the sessionId from /auth is treated like a bearer token that never expires, or a tight extraction loop trips the burst limit and the whole run dies on a 429 nobody planned for. Clinical data managers and Python ETL engineers hit these the moment a proof-of-concept script graduates to a scheduled sync against a live study. This page is the Veeva-specific integration primitive inside Deterministic Python ETL for EDC Data Extraction, within the broader Automated EDC Ingestion & Sync Pipelines discipline. It walks the full request lifecycle — pin the API version, authenticate for a session, extract with VQL, follow the next_page URL, and pace against the burst and daily limits — so every pull is resumable, hash-attested, and defensible under 21 CFR Part 11. The deterministic keyset approach to pagination itself lives in the companion Handling Pagination in Veeva Vault EDC APIs; this page is about wiring the integration around it.

The Vault CDMS Request Lifecycle at a Glance #

One extraction cycle authenticates once, issues a version-pinned VQL query, then follows the server-supplied next_page URL page by page while watching the X-VaultAPI limit headers, before landing and hashing each batch.

Root Cause: Why Vault CDMS Integrations Are Fragile #

Vault’s REST API is disciplined and well-documented, which lulls first integrations into three assumptions that production breaks.

The first is version drift. Every Vault REST call is pinned to a version segment in the path — /api/v25.1/, /api/v24.3/, and so on — and Vault runs several versions concurrently. Field names in the responseDetails envelope, default behaviors, and even VQL semantics can differ between versions. A script that hardcodes /api/v24.1/ keeps working after the tenant upgrades, right up until you rely on a field that moved; a script that omits the version entirely does not work at all. The version string is a configuration value under change control, not a constant to paste once.

The second is session lifetime. Authentication is a POST to /api/vXX.Y/auth with a username and password (or via OAuth / SAML for federated tenants); it returns a sessionId you place in the Authorization header of every subsequent call. That session is not eternal — it has an inactivity timeout and an absolute lifetime — so a long extraction that treats the id as permanent will start collecting 401s mid-run. The session must be refreshed proactively, exactly as a long Rave pull must, but through Vault’s own /auth contract.

The third is rate limiting you cannot see until you hit it. Vault enforces both a short-window burst limit and a rolling daily limit, and it advertises headroom on every response through the X-VaultAPI-BurstLimitRemaining and X-VaultAPI-DailyLimitRemaining headers. Ignore them and a tight next_page loop will exhaust the burst allowance and earn a 429; the run then dies with a fraction of the study extracted and an ambiguous cursor. Compounding all of this, VQL caps a single page at roughly 1,000 rows regardless of the PAGESIZE you request, so any real study must paginate — you never get the dataset in one call. The endpoint and envelope contract behind these calls is documented in EDC API Architecture for Clinical Trials.

Step-by-Step Implementation #

Each step is a single-responsibility building block; compose them into one Vault extraction worker per study.

1. Pin the API version in one place #

Treat the version segment as configuration, resolve it once, and build every URL from it. This is the cheapest defense against a silent post-upgrade break.

# 21 CFR Part 11 relevance: the pinned API version is a change-controlled config
# value; recording it makes every extracted record attributable to a known contract.
import os
import requests

VAULT_HOST = os.environ["VAULT_HOST"]              # e.g. my-sponsor.veevavault.com
VAULT_API_VERSION = os.environ["VAULT_API_VERSION"]  # e.g. v25.1 -- under change control

def api_base() -> str:
    # Never call bare /api/ -- the version segment is mandatory and audited.
    return f"https://{VAULT_HOST}/api/{VAULT_API_VERSION}"

2. Authenticate and hold a refreshable session #

POST to /auth, capture the sessionId, and wrap it so the worker can refresh proactively before Vault’s inactivity timeout rather than reacting to a mid-run 401.

# 21 CFR Part 11 relevance: authentication is an access-control event; we log the
# session rotation and principal, never the raw password, with a contemporaneous timestamp.
import time, logging

log = logging.getLogger("vault.session")

class VaultSession:
    """Holds a Vault sessionId, refreshed before its inactivity timeout."""

    def __init__(self, ttl_seconds: int = 600):
        self._ttl = ttl_seconds
        self._session_id: str | None = None
        self._issued_at = 0.0

    def token(self) -> str:
        if self._session_id is None or (time.monotonic() - self._issued_at) > self._ttl:
            self._authenticate()
        return self._session_id

    def _authenticate(self) -> None:
        resp = requests.post(
            f"{api_base()}/auth",
            data={"username": os.environ["VAULT_USER"], "password": os.environ["VAULT_PASSWORD"]},
            headers={"Accept": "application/json"},
            timeout=30,
        )
        resp.raise_for_status()
        body = resp.json()
        if body.get("responseStatus") != "SUCCESS":       # Vault signals failure in the body, not only the status code
            raise RuntimeError(f"Vault auth failed: {body.get('errors')}")
        self._session_id = body["sessionId"]
        self._issued_at = time.monotonic()
        log.info("vault_session_rotated", extra={"principal": os.environ["VAULT_USER"], "ts": time.time()})

3. Extract clinical data with a version-pinned VQL query #

Issue the query with the session id in the Authorization header. Pin an explicit ORDER BY so the result is deterministic across pages and re-runs, and select only the delta window you need.

# ALCOA+ requirement: Consistent + Complete -- a pinned ORDER BY and a bounded
# window make the VQL extraction deterministic and replayable to identical rows.
def run_vql(session: VaultSession, since_modified: str) -> dict:
    vql = (
        "SELECT id, subject__v, form__v, item_name__v, item_value__v, modified_date__v "
        "FROM edc_data_point__v "
        f"WHERE modified_date__v > '{since_modified}' "
        "ORDER BY modified_date__v ASC, id ASC "        # total order; survives ties across pages
        "PAGESIZE 1000"                                  # requested cap; Vault still enforces ~1000
    )
    resp = requests.post(
        f"{api_base()}/query",
        headers={"Authorization": session.token(), "Accept": "application/json"},
        data={"q": vql},
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()

4. Follow the next_page URL and pace against the limit headers #

Vault returns a relative responseDetails.next_page URL whenever more rows remain; follow it until it is absent. On every response, read the X-VaultAPI headers and slow down before the burst allowance is gone — treat a 429 as a failure to have paced, not a normal event.

# ALCOA+ requirement: Available -- pacing against the advertised burst headroom
# protects the Vault system of record from consumer-induced load (a Part 11 control).
def iter_pages(session: VaultSession, first: dict):
    body = first
    while True:
        yield body.get("data", [])
        details = body.get("responseDetails", {})
        next_url = details.get("next_page")             # relative path when more rows remain
        if not next_url:
            return                                       # dataset drained
        resp = requests.get(
            f"https://{VAULT_HOST}{next_url}",
            headers={"Authorization": session.token(), "Accept": "application/json"},
            timeout=60,
        )
        _pace(resp.headers)                              # slow down before the burst limit bites
        resp.raise_for_status()
        body = resp.json()

def _pace(headers) -> None:
    burst = int(headers.get("X-VaultAPI-BurstLimitRemaining", 1))
    daily = int(headers.get("X-VaultAPI-DailyLimitRemaining", 1))
    if burst <= 10:                                      # near the short-window ceiling
        time.sleep(2.0)
    if daily <= 100:                                     # protect the rolling daily budget
        log.warning("vault_daily_budget_low", extra={"remaining": daily})

5. Map, hash, and upsert each page idempotently #

Land each page as it arrives, stamp it with a SHA-256 over the raw rows before any transform, and upsert on the natural key so a replayed run produces identical state. The flattened rows then reconcile against CDISC ODM vs CDASH Schema Mapping.

# ALCOA+ requirement: Original + Attributable -- the batch hash fixes the exact
# bytes pulled, and the natural-key upsert makes a replayed page a no-op, not a dupe.
import hashlib, json

NATURAL_KEY = ("study", "subject", "form", "item", "record_id")

def to_row(rec: dict) -> dict:
    return {
        "study":     os.environ["VAULT_STUDY"],
        "subject":   rec["subject__v"],
        "form":      rec["form__v"],
        "item":      rec["item_name__v"],
        "record_id": rec["id"],
        "value":     rec.get("item_value__v"),
        "modified":  rec["modified_date__v"],
    }

def batch_hash(rows: list[dict]) -> str:
    canonical = json.dumps([[r[k] for k in (*NATURAL_KEY, "value")] for r in rows],
                           sort_keys=False, separators=(",", ":"), default=str)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

def land_page(dst, records: list[dict]) -> str:
    rows = [to_row(r) for r in records]
    digest = batch_hash(rows)
    with dst.begin():                                    # rows + audit committed together
        dst.upsert("edc_item_data", rows, on=NATURAL_KEY)
        dst.write_audit(batch_sha256=digest, record_count=len(rows))
    return digest

Verification and Audit Trail #

A Vault extractor is GxP-relevant software, so completeness must be provable from the log rather than asserted. Emit a structured record per page and a reconciliation per run, so an inspector can rebuild exactly which VQL window and which API version produced which landed batch.

Confirm the integration three ways against a mocked Vault endpoint: a session that crosses its TTL rotates before a 401 can land; following next_page to exhaustion yields a summed record_count equal to responseDetails.total for a quiesced window; and re-running the identical window reproduces the same per-page batch_sha256 with zero net upserts. The rate-limit pacing itself belongs to the parent Handling API Rate Limits in Clinical Sync guide, and the boundaries of what a read-only consumer may log follow Audit Trail Boundaries in EDC Systems. Discrepancies the extraction surfaces flow into Automated Clinical Query Generation.

Edge Cases and Vendor-Specific Gotchas #

Pin the API version, and version it like code. Because Vault serves several /api/vXX.Y versions at once, an unpinned or stale version segment is a silent time bomb — the field you read today may be renamed in the version you drift onto. Keep the version in configuration under change control, log it on every run, and treat a version bump as a validated change with a regression pass, not a one-line edit.

The VQL page cap is ~1,000 rows, no matter what you ask for. Requesting PAGESIZE 5000 does not get you 5,000 rows; Vault still returns at most about a thousand and hands you a next_page URL for the rest. Any code path that assumes a single response holds the whole result set will silently truncate a real study. Always follow next_page to exhaustion and reconcile the total.

Burst 429s carry their own headers — read them, don’t just retry. A 429 from the burst limit arrives with X-VaultAPI-BurstLimitRemaining at or near zero, and blindly retrying only deepens the hole. Back off on the burst window, and watch X-VaultAPI-DailyLimitRemaining so a large backfill does not consume the daily budget the rest of the day’s syncs depend on. The deterministic cursor discipline that makes those retries safe to resume is covered in Handling Pagination in Veeva Vault EDC APIs.

Frequently Asked Questions #

Related #

• Deterministic Python ETL for EDC Data Extraction — the parent extraction discipline this Vault integration plugs into.
• Handling API Rate Limits in Clinical Sync — pacing the burst and daily limits the X-VaultAPI headers advertise.
• Handling Pagination in Veeva Vault EDC APIs — the deterministic keyset cursor that makes next_page resumable across runs.
• Automating Medidata Rave Data Pulls with Python — the equivalent extraction pattern for a different EDC vendor.
• Automated EDC Ingestion & Sync Pipelines — the parent reference for this discipline.