A Prefect flow that loads a table and then reports Completed has told you the code ran. It hasn’t told you the data is right. Add one task that asks DataOps TestGen.
TestGen runs from Prefect through its REST API. No connector, and none needed. The Airflow answer and the CI/CD answer cover the same calls in those tools.
The calls
From Run Profiling and Tests, each with Authorization: Bearer <token>:
| Step | Call |
|---|---|
| Start a test run | POST /api/v1/test-suites/{test_suite_id}/test-runs |
| Start a profiling run | POST /api/v1/table-groups/{table_group_id}/profiling-runs |
| Poll either one | GET /api/v1/jobs/{job_id} until completed, error, or canceled |
| Read a test run | GET /api/v1/test-runs/{job_id} |
Submissions return 202 Accepted with a job id and don’t wait. So the task polls, with a timeout.
The task
import os, time, requests
from prefect import flow, task
API = os.environ["TESTGEN_API_URL"] # http://<host>:8530 by default
HDR = {"Authorization": f"Bearer {os.environ['TESTGEN_TOKEN']}"} # one-hour OAuth access token (or an Enterprise PAT)
class DataTestsFailed(Exception):
pass
@task(retries=2, retry_delay_seconds=60)
def testgen_run(suite_id: str, timeout_s: int = 1800) -> dict:
job_id = requests.post(f"{API}/api/v1/test-suites/{suite_id}/test-runs",
headers=HDR, timeout=30).json()["id"]
deadline = time.time() + timeout_s
while time.time() < deadline:
job = requests.get(f"{API}/api/v1/jobs/{job_id}", headers=HDR, timeout=30).json()
if job["status"] in ("completed", "error", "canceled"):
break
time.sleep(15)
else:
raise TimeoutError(f"TestGen job {job_id} did not finish")
if job["status"] != "completed":
raise RuntimeError(f"TestGen job {job_id}: {job['status']} {job['error_message']}")
return requests.get(f"{API}/api/v1/test-runs/{job_id}", headers=HDR, timeout=30).json()["result"]
@flow
def load_orders():
# ... your load tasks ...
result = testgen_run(os.environ["TG_SUITE_ORDERS"])
counts = result["result_counts"]
if counts["failed"] or counts["error"]:
raise DataTestsFailed(f"{counts['failed']} failed, score {result['score']}")
# ... downstream tasks run only past this line ...The check sits in the flow, not the task, so a retry never re-runs the verdict. Prefect’s states docs say an exception raised in the flow function marks the flow run Failed (checked October 2026). Retries cover the HTTP calls: a TestGen run that found bad data finds it again on a retry.
Test every load, profile every few weeks
Tests are cheap and belong after every load. Profiling reads every column, and a column’s shape changes slowly, so TestGen’s scheduling docs recommend it every two weeks to a month. Put profiling-runs in a separate flow on its own Prefect schedule, or skip Prefect for it and use TestGen’s scheduler.
What counts as a failure
TestGen tests carry a severity. One set to warn returns warning, not failed, per the testing docs. The flow above blocks on failed and error and lets warnings through. DataKitchen calls this a tripwire, and its medallion post puts one between each layer.
Where results land
In TestGen’s Test Results page. The test-run call takes an optional body with external_url and metadata; send the Prefect flow run URL and it appears under Link. The flow keeps the score and counts in its logs.
Open source authenticates with an OAuth 2.1 client. Access tokens last an hour and refresh tokens 30 days, so a flow that runs for a month needs a renewed refresh token, and renewing it takes a browser sign-in. Enterprise adds personal access tokens. There’s no Observability agent for Prefect, though TestGen can export results to Observability.
Documentation: TestGen API, Run Profiling and Tests, API Authentication, OAuth 2.1 Clients, Connect to Observability