End-to-End Guide

Test modes

Reconciliation

The engine function is snapline().

from snapline.engine import snapline

result = snapline(live_data, expected_data, {
    "ignoreFields": ["metadata.traceId"],
    "transformations": {
        "role": lambda v, _k, _p: str(v).upper(),
    },
    "dataMapping": {
        "status": {"synced": "SYNCED"},
    },
})
print(result["match"], result.get("diff"))

API vs file

REST

import asyncio
from snapline.core import auth, test_suite

async def main():
    await test_suite("REST snapshot", {
        "baseUrl": "https://api.example.com",
        "auth": auth.oauth2({
            "tokenUrl": "https://api.example.com/oauth/token",
            "clientId": "id", "clientSecret": "secret",
        }),
        "api": {
            "endpoint": "/api/v1/user/sync",
            "method": "POST",
            "inputFile": "fixtures/input.json",
            "expectedFile": "fixtures/expected.json",
        },
    })

asyncio.run(main())

GraphQL

from snapline.core import api, test_suite

await test_suite("GraphQL snapshot", {
    "baseUrl": "https://api.example.com",
    "api": {
        **api.graphql({
            "endpoint": "/graphql",
            "queryFile": "fixtures/query.graphql",
            "variablesFile": "fixtures/variables.json",
            "dataPath": "customerAccount",
        }),
        "expectedFile": "fixtures/expected.json",
    },
})

SOAP

await test_suite("SOAP snapshot", {
    "baseUrl": "https://api.example.com",
    "api": {
        **api.soap({
            "endpoint": "/soap/user",
            "soapAction": "GetUser",
            "inputFile": "fixtures/request.xml",
        }),
        "expectedFile": "fixtures/expected.json",
    },
})

Database comparison

Implement DbConnectionLike with your SQL driver, or use the in-memory example from packages/snapline_core/examples/in_memory_db.py.

from snapline.core.examples.in_memory_db import InMemoryDb
from snapline.core import test_suite

source_db = InMemoryDb([{"status": "ABC", "email": "alice@example.com"}])
target_db = InMemoryDb([{"status": "CBA", "email": "alice@example.com"}])

await test_suite("Warehouse sync", {
    "dbComparison": {
        "sourceDb": source_db,
        "targetDb": target_db,
        "query": "SELECT status, email FROM users WHERE email = :email",
        "params": {"email": "alice@example.com"},
        "dataMapping": {"status": {"ABC": "CBA"}},
    },
})

Cross-system

from snapline.core import api, test_suite
from snapline.core.examples.in_memory_db import InMemoryDb

app_db = InMemoryDb([{"email": "alice@example.com", "status": "SYNCED"}])

await test_suite("API matches DB", {
    "baseUrl": "https://api.example.com",
    "apiToDb": {
        "api": api.rest({
            "endpoint": "/users/profile?email=alice@example.com",
            "method": "GET",
        }),
        "db": {
            "db": app_db,
            "query": "SELECT email, status FROM users WHERE email = :email",
            "params": {"email": "alice@example.com"},
        },
    },
})

Warehouse comparison (SQL → NoSQL)

For multi-table ETL into a document store, use run_warehouse_comparison — see demo project-db.

from snapline.core import nosql, run_warehouse_comparison

await run_warehouse_comparison({
    "suiteName": "Warehouse sync",
    "sourceDb": source_db,
    "targetDb": nosql.memory(),
    "tables": [{
        "id": "wh_customers",
        "sourceQuery": "SELECT customer_id AS customerId, email FROM wh_customers",
        "targetCollection": "customers",
        "linkKeys": {"customerId": "customerId"},
    }],
    "chunkSize": 50,
    "report": {"outputPath": "./reports/warehouse.jsonl", "format": "jsonl"},
})

Queue → poll (async pipelines)

Install snapline-messaging-adapters for in-memory or custom queue publishers. Use publishAndPoll to wait for results in SQL or on disk:

from snapline.messaging_adapters import messaging
from snapline.core import run_publish_and_poll

queue = messaging.memory()

result = await run_publish_and_poll({
    "publish": {"publisher": queue["publisher"], "topic": "jobs.submit", "payload": {"jobId": "JOB-42"}},
    "poll": {"file": {"directory": "/var/results"}},
    "expected": {"jobId": "JOB-42", "status": "COMPLETE"},
})

Fixture cases

import asyncio
from snapline.core import fixtures_dir, run_api_fixture_cases

async def main():
    result = await run_api_fixture_cases({
        "suiteName": "GraphQL fixture cases",
        "fixturesRoot": str(fixtures_dir(__file__, relative_path="../fixtures")),
        "baseUrl": os.environ["API_BASE_URL"],
        "defaults": {
            "endpoint": "/graphql",
            "protocol": "graphql",
            "dataPath": "customerAccount",
        },
    })
    raise SystemExit(0 if result["passed"] else 1)

Offline cases:

from snapline.core import run_snapline_fixture_cases

await run_snapline_fixture_cases({
    "suiteName": "Transformations — offline",
    "fixturesRoot": str(fixtures_dir(__file__)),
})

Reports

from snapline.core import resolve_report_config, write_test_report

result = await test_suite("User sync", {...})
config = resolve_report_config({"defaultOutputPath": "reports/run.json"})
if config:
    write_test_report([result], config)

REPORT_FORMAT=html REPORT_OUTPUT=reports/demo.html uv run demo
REPORT_REDACT_FIELDS=data,processed uv run demo

Snapline Hub (optional)

Push reports to Snapline Hub for a centralized dashboard. Hub is optional — file-based reports above work without it.

from snapline.core import build_report, push_test_report_to_hub

report = build_report([result], {"durationMs": 1200})
push_test_report_to_hub(report, hub_url="http://localhost:3847")

# Or via env:
# SNAPLINE_HUB_URL=http://localhost:3847 uv run demo

Scenario modules (demo pattern)

Each demo scenario is a scenario.py with a Scenario class:

class Scenario:
    id = "api-vs-file-rest"
    name = "API vs file (REST + OAuth2)"
    needs_server = True

    async def run(self, ctx):
        return await test_suite(self.name, {...})

Copy demo/scenarios/<id>/ into your project as a template.