Healthcare integration on PostgreSQL

Status

runnable in repo test

Industry

Healthcare

Language

Python

Framework

pytest

CI

any

Fulfilment

BUILD_DATASET

Profile

healthcare-integration

About this example

All data is synthetic. DataNivra provisions approved test data into a customer-controlled target; it does not run these tests. Credentials for the target are never returned by DataNivra — your runner holds its own.

Why test data matters

Integration tests for scheduling, claims and lab workflows need many related records — members with coverage, encounters, claims and lab results — without ever touching real patient data.

What is requested

The approved healthcare-integration@1 profile with its defaults: HC_ROUTINE_CARE × 500 and HC_CHRONIC_CARE × 2000 (2,500 records in total).

Policy gates

Profile version approved by a second person; synthetic-only build (no source is read); counts must stay within the profile maximums (500 / 2,000; ceiling 3,000); the automation envelope allows 2,500 records for the named CI workload identity.

What is provisioned

A certified synthetic dataset built by the customer's agent into the profile's approved PostgreSQL target environment.

Expected output

Cleanup

TestDataClient.acquire releases the request; releasing a build revokes the dataset that was built for this request alone.

How the promptbook illustration maps to the real healthcare pack

The design brief spoke of "500 patients and 2,000 appointments". The healthcare pack has no Patient or Appointment entity: a member is the patient and an encounter is the visit (appointment). Scenario counts are exact numbers of member root entities; encounters, claims and labs follow from each scenario's fan-out. This example therefore requests 500 routine-care members and 2,000 chronic-care members (each with several encounters). It does not promise an exact number of encounters.

Files

• request.yaml
• example.py
• test_with_datanivra.py

Run

1. A tenant administrator defines and a second person approves the profile (datanivra test-data define ../../profiles/healthcare-integration.yaml --submit, then datanivra policies approve).
2. datanivra test-data validate request.yaml && datanivra test-data plan request.yaml
3. Run the example with DATANIVRA_BASE_URL and a session or workload token.

# yaml-language-server: $schema=../../../../packages/contracts/schemas/v1/manifests/test-data-request.schema.json
# DataNivra test-data-request manifest (apiVersion datanivra.io/v1alpha1). Validate: datanivra test-data validate <file>
apiVersion: datanivra.io/v1alpha1
kind: TestDataRequest
metadata:
  name: healthcare-integration-pr
spec:
  profile: { name: healthcare-integration, version: 1 }
  purpose_ref: QA-1234

"""Healthcare integration on PostgreSQL — provision approved test data through the Python SDK.

``provision`` only asks DataNivra for data from the approved ``healthcare-integration`` profile; your own test
suite runs against the approved target with its own, separately managed database access.
"""

from __future__ import annotations

from pathlib import Path

from datanivra_contracts.testdata import TestDataRequest
from datanivra_sdk import DataNivraClient
from datanivra_sdk.testdata import TestDataClient

REQUEST = Path(__file__).with_name("request.yaml")


def provision(api: DataNivraClient, *, idempotency_key: str | None = None) -> TestDataRequest:
    """Create (or, for the same key, return) the request; DENIED raises ``DataNivraApiError``."""
    return TestDataClient(api).request(REQUEST, idempotency_key=idempotency_key)

"""Healthcare integration on PostgreSQL (pytest).

Run in your CI with ``DATANIVRA_BASE_URL`` and a token (``datanivra login --workload-identity ...``
or ``DATANIVRA_TOKEN``). The data is released after the session, even when tests fail.
"""

from __future__ import annotations

import os
from collections.abc import Iterator

import pytest
from datanivra_contracts.testdata import TestDataRequest
from datanivra_sdk import DataNivraClient
from datanivra_sdk.testdata import TestDataClient
from example import REQUEST


@pytest.fixture(scope="session")
def test_data() -> Iterator[TestDataRequest]:
    api = DataNivraClient(os.environ["DATANIVRA_BASE_URL"], token=os.environ["DATANIVRA_TOKEN"])
    with TestDataClient(api).acquire(REQUEST, idempotency_key=os.environ.get("CI_RUN_KEY")) as request:
        yield request


def test_members_and_encounters_are_available(test_data: TestDataRequest) -> None:
    # Your assertions run against your application and the approved target environment
    # ``test_data.environment_id``; DataNivra only returns metadata (ids, states, digests).
    assert test_data.state.value == "READY"

# yaml-language-server: $schema=../../../packages/contracts/schemas/v1/manifests/test-data-profile.schema.json
# DataNivra test-data-profile manifest (apiVersion datanivra.io/v1alpha1). Validate: datanivra test-data validate <file>
# Placeholder ids (00000000-0000-4000-8000-0000000000NN): replace with your ids.
# 01 = approved target environment, 11 = certified dataset, 21 = customer agent,
# 99 = workload identity trust (datanivra workload-identities list).
apiVersion: datanivra.io/v1alpha1
kind: TestDataProfile
metadata:
  name: healthcare-integration
  description: Synthetic members (patients) and their encounters (appointments) for integration tests
spec:
  fulfilment: BUILD_DATASET
  build:
    business_domain: patient encounters
    pack_id: healthcare
    dataset_kind: SYNTHETIC_ONLY
    target_environment_id: 00000000-0000-4000-8000-000000000001
    retention_days: 1
    max_total_records: 3000
    scenarios:
      - {
          pack_id: healthcare,
          scenario_code: HC_ROUTINE_CARE,
          kind: NORMAL,
          default_count: 500,
          max_count: 500,
        }
      - {
          pack_id: healthcare,
          scenario_code: HC_CHRONIC_CARE,
          kind: HIGH_VOLUME,
          default_count: 2000,
          max_count: 2000,
        }
      - {
          pack_id: healthcare,
          scenario_code: HC_CLAIM_DENIALS,
          kind: RARE,
          default_count: 50,
          max_count: 400,
          included_by_default: false,
        }
      - {
          pack_id: healthcare,
          scenario_code: HC_ABNORMAL_LABS,
          kind: BOUNDARY,
          default_count: 50,
          max_count: 200,
          included_by_default: false,
        }
  automation:
    workload_identity_ids: [00000000-0000-4000-8000-000000000099]
    max_total_records: 2500
    max_live_requests: 10
    max_requests_per_hour: 60
    escalate_out_of_envelope: true
  audit_tags: [HEALTHCARE_SYNTHETIC]