Query Completion Data with Python

This workflow queries completion wells through the Platform API and reads one-second corva#completion.wits records through the Data API.

Find completion wells

import os
import requests

headers = {"Authorization": f"API {os.environ['CORVA_API_KEY']}"}

response = requests.get(
    "https://api.corva.ai/v2/wells",
    headers=headers,
    params={
        "visibility": "visible",
        "status[]": ["active", "paused", "complete"],
        "state[]": ["completions", "completed"],
        "fields[]": ["well.name", "well.asset_id", "well.state", "well.status"],
        "sort": "name",
        "order": "asc",
        "per_page": 1000,
    },
    timeout=30,
)
response.raise_for_status()
wells = response.json()["data"]

for well in wells:
    attributes = well["attributes"]
    print(attributes["name"], attributes["asset_id"])

Read completion WITS

import json

asset_id = 12345
cursor = 0
records = []

while True:
    response = requests.get(
        "https://data.corva.ai/api/v1/data/corva/completion.wits/",
        headers=headers,
        params={
            "query": json.dumps({
                "asset_id": asset_id,
                "timestamp": {"$gt": cursor},
            }),
            "sort": json.dumps({"timestamp": 1}),
            "limit": 10000,
            "fields": (
                "timestamp,asset_id,stage_number,data.wellhead_pressure,"
                "data.slurry_flow_rate_in,data.bottomhole_pressure,"
                "data.proppant_1_concentration"
            ),
        },
        timeout=30,
    )
    response.raise_for_status()
    batch = response.json()

    if not batch:
        break

    records.extend(batch)
    cursor = batch[-1]["timestamp"]

print(f"Retrieved {len(records)} records")

Dataset fields can vary by operation and permissions. Confirm the required fields in the Dataset Explorer before building a fixed export schema.