#!/usr/bin/env python3 """Run the Entropy Data API usage examples end to end. This script executes the workflow shown at https://docs.entropy-data.com/usage-examples against an Entropy Data instance: it registers a team, a data contract, a data product, a semantic concept, an access with provisioning reports, an asset, example data, and test results, verifies each resource with a GET, and cleans everything up afterwards. Requirements: Python 3.8+, no third-party packages. Usage: export ENTROPY_DATA_API_KEY=your-secret-api-key python3 usage-examples.py [--host https://api.entropy-data.com] [--prefix test-] [--keep] [--force] [--integration your-integration-external-id] The API key must have organization scope, so the script can create teams. Use --prefix to namespace all created ids (recommended), --keep to skip the cleanup, and --force to overwrite resources that already exist. """ import argparse import json import os import sys import urllib.error import urllib.request import uuid from datetime import datetime, timedelta, timezone results = [] def request(method, path, body=None, expected=(200,)): """Send a request and return (status, parsed JSON or None).""" url = ARGS.host.rstrip("/") + path data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("x-api-key", API_KEY) if data is not None: req.add_header("content-type", "application/json") try: with urllib.request.urlopen(req) as response: raw = response.read() status = response.status except urllib.error.HTTPError as error: raw = error.read() status = error.code try: parsed = json.loads(raw) if raw else None except json.JSONDecodeError: parsed = None if expected and status not in expected: detail = "" if isinstance(parsed, dict): detail = " " + str(parsed.get("detail") or parsed.get("title") or "") raise RuntimeError(f"{method} {path} returned {status}{detail}") return status, parsed def step(name, func): """Run one step and record the outcome.""" try: outcome = func() results.append((name, outcome or "PASS")) print(f" [{outcome or 'PASS':<4}] {name}") except RuntimeError as error: results.append((name, "FAIL")) print(f" [FAIL] {name}: {error}") if not ARGS.keep_going: raise def guard_free(path): """Abort if the resource already exists, so nothing real is overwritten.""" status, _ = request("GET", path, expected=None) if status == 200 and not ARGS.force: raise RuntimeError( f"{path} already exists; use --force to overwrite or --prefix to rename" ) def put_and_verify(path, body): guard_free(path) request("PUT", path, body) request("GET", path) # ---------------------------------------------------------------- workflow def create_team(team_id, name): put_and_verify(f"/api/teams/{team_id}", { "id": team_id, "name": name, "type": "Team", "description": f"This is the {name.lower()} team", "members": [], }) def create_data_contract(): put_and_verify(f"/api/datacontracts/{CONTRACT_ID}", { "apiVersion": "v3.1.0", "kind": "DataContract", "id": CONTRACT_ID, "name": "Customer Cohorts", "version": "1.0.0", "status": "active", "description": { "purpose": "A table with customer cohorts and their properties", "usage": "Max. 10x queries per day", "limitations": "Not suitable for real-time use cases", }, "servers": [{ "server": "production", "type": "snowflake", "account": "lmtrwxs-xn14859", "database": "MARKETING_DB", "schema": "CUSTOMER_COHORTS", }], "schema": [{ "name": "customer_cohorts", "physicalType": "table", "properties": [ {"name": "customer_id", "logicalType": "string", "description": "Unique customer identifier.", "required": True, "primaryKey": True}, {"name": "cohort_name", "logicalType": "string", "description": "Name of the cohort."}, ], }], "team": {"name": TEAM_ID}, }) def create_data_product(): put_and_verify(f"/api/dataproducts/{PRODUCT_ID}", { "apiVersion": "v1.0.0", "kind": "DataProduct", "id": PRODUCT_ID, "name": "Customer Cohorts", "status": "active", "description": { "purpose": "Customer cohorts and their properties for campaign targeting", }, "outputPorts": [{ "name": OUTPUT_PORT_ID, "version": "1", "description": "Customer cohorts table in Snowflake", "type": "snowflake", "contractId": CONTRACT_ID, "customProperties": [ {"property": "status", "value": "active"}, {"property": "containsPii", "value": False}, {"property": "server", "value": { "account": "lmtrwxs-xn14859", "database": "MARKETING_DB", "schema": "CUSTOMER_COHORTS", }}, ], }], "team": {"name": TEAM_ID}, }) def create_semantic_concept(): guard_free(f"/api/semantics/experimental/namespaces/{NAMESPACE}") request("PUT", f"/api/semantics/experimental/namespaces/{NAMESPACE}", { "namespace": NAMESPACE, "name": "Main", "description": "The default namespace for the organization.", }) path = f"/api/semantics/experimental/namespaces/{NAMESPACE}/concepts/customer" request("PUT", path, { "id": "customer", "name": "Customer", "kind": "entity", "description": "A person or company that has purchased at least once.", "status": "active", }) request("GET", path) def create_and_approve_access(): body = { "id": ACCESS_ID, "info": { "purpose": "Verify the API usage examples end to end", "startDate": datetime.now(timezone.utc).date().isoformat(), "endDate": (datetime.now(timezone.utc) + timedelta(days=1)).date().isoformat(), }, "provider": { "dataProductId": PRODUCT_ID, "outputPortId": OUTPUT_PORT_ID, }, "consumer": {"teamId": CONSUMER_TEAM_ID}, } put_and_verify(f"/api/access/{ACCESS_ID}", body) request("POST", f"/api/access/{ACCESS_ID}/approve") def report_provisioning(): request("POST", f"/api/access/{ACCESS_ID}/provisioning/provisioning-succeeded", { "platform": "snowflake", "executor": "usage-examples.py", "reference": "CUSTOMER_COHORTS_V1_READER", }) def create_asset(): put_and_verify(f"/api/assets/{ASSET_ID}", { "id": ASSET_ID, "info": { "source": "snowflake", "sourceId": ASSET_ID, "type": "snowflake_table", "name": "customer_cohorts", "status": "active", "description": "Customer cohorts and their properties.", }, "columns": [ {"name": "customer_id", "type": "string", "description": "Unique customer identifier", "required": True, "primaryKey": True}, {"name": "cohort_name", "type": "string", "description": "Name of the cohort"}, ], }) def trigger_integration_run(): if not ARGS.integration: return "SKIP" status, _ = request( "POST", f"/api/integrations/{ARGS.integration}/run", expected=(202, 409) ) return "PASS" if status == 202 else "SKIP" def create_example_data(): put_and_verify(f"/api/example-data/{EXAMPLE_DATA_ID}", { "id": EXAMPLE_DATA_ID, "dataProductId": PRODUCT_ID, "outputPortId": OUTPUT_PORT_ID, "schemaName": "customer_cohorts", "data": [ {"CUSTOMER_ID": "CUST-4821", "COHORT_NAME": "high-value"}, {"CUSTOMER_ID": "CUST-7733", "COHORT_NAME": "churn-risk"}, ], }) def publish_test_results(): now = datetime.now(timezone.utc) request("POST", "/api/test-results", { "id": TEST_RESULT_ID, "dataContractId": CONTRACT_ID, "dataContractVersion": "1.0.0", "server": "production", "timestampStart": (now - timedelta(seconds=14)).isoformat(), "timestampEnd": now.isoformat(), "result": "passed", "checks": [{ "type": "schema", "name": "Check that field customer_id is present", "model": "customer_cohorts", "field": "customer_id", "result": "passed", "engine": "usage-examples.py", }], }) request("GET", f"/api/test-results/{TEST_RESULT_ID}") def cleanup(): """Delete everything the script created, in reverse order.""" deletions = [ f"/api/test-results/{TEST_RESULT_ID}", f"/api/example-data/{EXAMPLE_DATA_ID}", f"/api/assets/{ASSET_ID}", f"/api/access/{ACCESS_ID}", f"/api/dataproducts/{PRODUCT_ID}", f"/api/datacontracts/{CONTRACT_ID}", f"/api/semantics/experimental/namespaces/{NAMESPACE}/concepts/customer", f"/api/semantics/experimental/namespaces/{NAMESPACE}", f"/api/teams/{CONSUMER_TEAM_ID}", f"/api/teams/{TEAM_ID}", ] for path in deletions: status, _ = request("DELETE", path, expected=None) label = "OK" if status in (200, 204) else status print(f" [{label:>4}] DELETE {path}") # ---------------------------------------------------------------- main parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--host", default=os.environ.get( "ENTROPY_DATA_HOST", "https://api.entropy-data.com")) parser.add_argument("--prefix", default="", help="prefix for all created ids, e.g. 'test-'") parser.add_argument("--keep", action="store_true", help="keep the created resources instead of cleaning up") parser.add_argument("--force", action="store_true", help="overwrite resources that already exist") parser.add_argument("--keep-going", action="store_true", help="continue with the next step after a failure") parser.add_argument("--integration", default="", help="externalId of an integration to trigger a run for") ARGS = parser.parse_args() API_KEY = os.environ.get("ENTROPY_DATA_API_KEY") if not API_KEY: sys.exit("Set the ENTROPY_DATA_API_KEY environment variable first.") TEAM_ID = ARGS.prefix + "marketing" NAMESPACE = ARGS.prefix + "main" CONSUMER_TEAM_ID = ARGS.prefix + "sales" CONTRACT_ID = ARGS.prefix + "customer-cohorts" PRODUCT_ID = ARGS.prefix + "customer-cohorts" OUTPUT_PORT_ID = "snowflake-customer-cohorts-v1" ACCESS_ID = str(uuid.uuid4()) ASSET_ID = ARGS.prefix + "customer-cohorts-table" EXAMPLE_DATA_ID = ARGS.prefix + "customer-cohorts-example" TEST_RESULT_ID = str(uuid.uuid4()) print(f"Running the usage examples against {ARGS.host}") try: step("Set up teams", lambda: create_team(TEAM_ID, "Marketing")) step("Set up consumer team", lambda: create_team(CONSUMER_TEAM_ID, "Sales")) step("Register a data contract", create_data_contract) step("Register a data product", create_data_product) step("Model your semantics", create_semantic_concept) step("Automate access decisions", create_and_approve_access) step("Report provisioning results", report_provisioning) step("Ingest assets", create_asset) step("Trigger an ingestion run", trigger_integration_run) step("Attach example data", create_example_data) step("Publish test results", publish_test_results) except RuntimeError: pass # the failed step is already recorded, stop and clean up finally: if not ARGS.keep: print("Cleaning up:") cleanup() failed = [name for name, outcome in results if outcome == "FAIL"] print(f"\n{len(results) - len(failed)}/{len(results)} steps passed" + (f", failed: {', '.join(failed)}" if failed else "")) sys.exit(1 if failed else 0)