API Reference

Quick reference for snapline.core. Config dict keys use camelCase for parity with Node.js.

Core orchestration

Function	Description
`await test_suite(name, config)`	Run test modes; returns dict with `passed`, `results`
`await run_api_fixture_cases(options)`	HTTP fixture cases
`await run_snapline_fixture_cases(options)`	Offline fixture cases

Reconciliation

Function	Description
`snapline(live, expected, options)`	Full reconciliation pipeline
`assert_against_file(live, path, options)`	Compare live data vs JSON file
`load_json_file(path, root_dir=None)`	Load JSON with optional sandboxing

API factories

from snapline.core import api

api.rest({...})
api.graphql({...})
api.soap({...})

Database connection contract

Published packages do not ship SQL drivers. Implement DbConnectionLike with your Postgres/MySQL client:

from snapline.core.types import DbConnectionLike

class AppDb(DbConnectionLike):
    async def query(self, sql: str, params: dict | None = None) -> list[dict]:
        # your pg/mysql driver here
        return rows

Demos use snapline.demo_shared for SQLite and in-memory stubs. See packages/snapline_core/examples/in_memory_db.py for a minimal stub you can copy.

Queue & async polling

Function	Description
`await run_publish_and_poll(config)`	Publish message, poll async result, compare
`create_db_async_result_resolver(config)`	Poll SQL until rows ready
`create_file_async_result_resolver(config)`	Poll JSON file on disk
`wait_for_result(check, ...)`	Generic poll helper

from snapline.messaging_adapters import messaging
from snapline.core import test_suite

queue = messaging.memory()
publisher = queue["publisher"]

await test_suite("Order pipeline", {
    "publishAndPoll": {
        "publish": {"publisher": publisher, "topic": "orders.request", "payload": {"orderId": "ORD-1"}},
        "poll": {"db": {"db": app_db, "query": "SELECT status FROM orders WHERE correlationId = :correlationId"}},
        "expected": {"status": "PROCESSED"},
    },
})

Warehouse comparison

from snapline.core import run_warehouse_comparison, nosql, create_stream_report_writer

await run_warehouse_comparison({
    "sourceDb": source_db,           # DbConnectionLike — your SQL warehouse
    "targetStore": nosql.memory(),   # DocumentStoreLike — e.g. MongoDB adapter
    "tables": [...],                 # WarehouseTableSpec list
    "chunkSize": 50,
    "report": {"outputPath": "./reports/warehouse.jsonl"},
})

create_stream_report_writer appends JSONL events without holding the full dataset in memory.

NoSQL

from snapline.core import nosql

store = nosql.memory()

Auth

from snapline.core import auth

auth.oauth2({...})
auth.openid({...})
auth.basic({...})

Reporting

Function	Description
`resolve_report_config(options)`	Parse CLI / env
`write_test_report(suites, config, meta=None)`	Write report file
`push_test_report_to_hub(report, hub_url=...)`	Push report to Snapline Hub
`resolve_hub_config(argv=None)`	Parse `SNAPLINE_HUB_URL` / `--hub-url`
`redact_fields(data, fields)`	Mask sensitive values
`redact_suite_results(suites, fields)`	Redact all step results

Fixture utilities

from snapline.core import fixtures_dir, module_dir

fixtures_dir(__file__, relative_path="../fixtures")
module_dir(__file__)

Package examples

packages/snapline_core/examples/ — basic_usage, api_to_db, db_to_api, db_comparison, graphql, soap, nosql
packages/snapline_engine/examples/basic_snapline.py
packages/snapline_api_adapters/examples/
packages/snapline_auth_adapters/examples/basic_auth.py

pytest

pytest tests -q

Unit tests cover snapline, safe_path, safe_url, redact_fields, and xml_utils.