datatoolpack 0.7.2

Official Python SDK for the AutoData ML data preparation pipeline API

AutoData Python Client

Official Python SDK for the AutoData ML data preparation pipeline API.

Installation

pip install datatoolpack

Or install from source:

git clone https://github.com/datatoolpack/datatoolpack
cd datatoolpack
pip install .

Quick Start

from datatoolpack import AutoDataClient

with AutoDataClient(api_key="dtpk_YOUR_API_KEY") as client:
    result = client.process(
        file_path="data.csv",
        target_columns=["price"],
        output_rows=20000,
    )
    print(result["files"])

Get your API key from the AutoData dashboard → API Keys tab.

---

Authentication

API Key (recommended)

with AutoDataClient(api_key="dtpk_YOUR_API_KEY") as client:
    ...

Access Code (passcode)

If you have an access code instead of an API key:

with AutoDataClient(passcode="123456789012") as client:
    ...

Environment Variables

Set environment variables to avoid hardcoding credentials:

export AUTODATA_API_KEY="dtpk_YOUR_API_KEY"
# or
export AUTODATA_PASSCODE="123456789012"
# optional
export AUTODATA_BASE_URL="https://autodata.datatoolpack.com"

# No need to pass api_key — reads from AUTODATA_API_KEY automatically
with AutoDataClient() as client:
    result = client.process("data.csv", target_columns="price")

---

Supported File Formats

Format	Extensions
CSV	.csv
Excel	.xlsx, .xls
Parquet	.parquet
JSON	.json
Feather	.feather
ORC	.orc

---

Reference

AutoDataClient(api_key, passcode, base_url, timeout, max_retries)

Parameter	Type	Default	Env Variable	Description
api_key	str	None	AUTODATA_API_KEY	API key starting with dtpk_
passcode	str	None	AUTODATA_PASSCODE	Access code (alternative to api_key)
base_url	str	"https://autodata.datatoolpack.com"	AUTODATA_BASE_URL	Server URL (no trailing slash)
timeout	int	120		Request timeout in seconds
max_retries	int	3		Auto-retries for 429 / 502 / 503 / 504

Either api_key or passcode is required (via argument or env variable).

Implements context manager — use with with to auto-close connections:

with AutoDataClient(api_key="dtpk_...") as client:
    ...
# session is closed automatically

Or close manually:

client = AutoDataClient(api_key="dtpk_...")
# ... use client ...
client.close()

---

Core Pipeline

client.process(...) — Upload & run pipeline

result = client.process(
    file_path="data.csv",           # Path to input file (CSV, XLSX, Parquet, ...)
    target_columns=["price"],       # y-column(s) for ML
    output_rows=20000,              # Target row count in output
    tools={                         # Toggle pipeline steps (all optional)
        "anomaly": False,           # Anomaly detection (off by default)
        "dtc": True,                # Data Type Conversion
        "mdh": True,                # Missing Data Handler
        "dor": False,               # Dimensionality Reduction (off by default)
        "cds": True,                # Column Scaling
        "dsm": True,                # Data Split Manager
        "dsg": True,                # Synthetic Data Generator
    },
    advanced_params={               # Fine-grained parameters (all optional)
        "excluded_columns": ["id"], # Columns to drop before processing
        "text_mode": 0,             # 0=none, 1=neural, 2=tfidf
        "text_cleaning": True,      # Clean text before encoding
        "zscore_limit": 3.0,        # Z-score outlier threshold
        "dsg_mode": "copula",       # "copula" or "gan"
        "similarity_p": 95,         # Similarity percentile for DSG
    },
    wait=True,                      # Block until complete (default True)
    poll_interval=2,                # Status poll interval in seconds
    download_path="./outputs/",     # Where to save files (default auto)
    auto_download=True,             # Set False to skip download when wait=True
    output_preferences=["dsg.csv"], # Which files to download (default all)
    compressed=True,                # Download as ZIP (default True)
)

Returns a dict:

{
    "session_id": "abc123...",
    "status": "completed",
    "files": [
        {"name": "dsg.csv", "url": "/download/.../dsg.csv", "size": 2097152, "description": "..."},
        {"name": "dsm_train.csv", ...},
    ],
    "row_count": 20000,
    "duration_seconds": 42.1,
}

Set wait=False to get back immediately with just session_id and status:

result = client.process(file_path="data.csv", target_columns="price", wait=False)
session_id = result["session_id"]

---

client.get_status(session_id) — Poll progress

status = client.get_status(session_id)
# {
#   "status": "running",           # queued | running | completed | error | cancelled
#   "message": "Running MDH...",
#   "current_step": 3,
#   "total_steps": 6,
#   "progress_percent": 50,
#   "duration_seconds": 15.3,
# }

---

client.get_result(session_id) — Fetch completed results

result = client.get_result(session_id)
# {"status": "completed", "files": [...], "row_count": ..., "duration_seconds": ...}

---

client.wait_for_completion(session_id, poll_interval) — Block until done

result = client.wait_for_completion(session_id, poll_interval=3)

Prints live progress to stdout. Raises AutoDataError if processing fails.

---

client.cancel(session_id) — Cancel a running job

cancelled = client.cancel(session_id)  # True if acknowledged

---

client.retry_session(session_id) — Retry a failed pipeline

Retry from the last checkpoint — no need to re-upload or re-configure:

result = client.retry_session(session_id="failed-session-id")
# {'success': True, 'session_id': '...', 'message': 'Retrying from last checkpoint'}

---

client.download_results(session_id, ...) — Download output files

path = client.download_results(
    session_id,
    download_path="./my_outputs/",      # Directory to save into
    output_preferences=["dsg.csv"],     # Specific files only (None = all)
    compressed=True,                    # ZIP download (default) or individual files
)
print(f"Saved to {path}")

---

client.download_file(url, output_path) — Download a single file

client.download_file("/api/v1/download/abc123.../dsg.csv", "dsg.csv")

---

API Keys & Usage

client.list_keys() — List API keys

keys = client.list_keys()
# [{"id": "...", "name": "My Key", "prefix": "dtpk_abc123", "created_at": "..."}]

client.get_usage() — Usage statistics

usage = client.get_usage()
# {
#   "daily_credits_used": 500,
#   "daily_credit_limit": 10000,
#   "daily_remaining": 9500,
#   "lifetime_credits_used": 12340,
#   "lifetime_credit_limit": 1000000,
#   "lifetime_remaining": 987660,
#   "daily_request_count": 3,
#   "last_used_at": "2026-04-12T10:30:00Z",
# }

---

Connectors

Read data directly from databases and cloud storage instead of uploading files.

Supported connector types

Type	Description	Required secrets
sql	PostgreSQL, MySQL, SQL Server	connection_string
snowflake	Snowflake Data Cloud	account, user, password, database, schema, warehouse
bigquery	Google BigQuery	credentials_json, project_id, dataset
mongodb	MongoDB / Atlas	connection_string, database
s3	Amazon S3	bucket, access_key_id, secret_access_key, region
gcs	Google Cloud Storage	bucket, credentials_json
databricks	Databricks Lakehouse	host, token, catalog, schema
delta	Delta Lake	path (+ cloud credentials if remote)
fabric	Microsoft Fabric	workspace_id, lakehouse_id, tenant_id, client_id, client_secret
kafka	Apache Kafka	bootstrap_servers, topic, group_id
kinesis	Amazon Kinesis	stream_name, region, access_key_id, secret_access_key

Quick connector example

with AutoDataClient(api_key="dtpk_...") as client:
    # Test the connection
    client.test_connector("sql", secrets={
        "connection_string": "postgresql://user:pass@host/db"
    })

    # List tables
    tables = client.discover("sql", secrets={
        "connection_string": "postgresql://user:pass@host/db"
    })
    print(tables)  # ["users", "orders", ...]

    # Preview columns
    info = client.preview("sql", table="orders", secrets={
        "connection_string": "postgresql://user:pass@host/db"
    })
    print(info["columns"])  # ["id", "price", "date", ...]

    # Run the full pipeline from the connector
    result = client.process_from_connector(
        connector_type="sql",
        table="orders",
        target_columns=["price"],
        secrets={"connection_string": "postgresql://user:pass@host/db"},
        output_rows=20000,
    )
    print(result["files"])

Connector examples by type

Snowflake:

secrets = {
    "account": "abc123.us-east-1",
    "user": "analyst",
    "password": "...",
    "database": "PROD",
    "schema": "PUBLIC",
    "warehouse": "COMPUTE_WH",
}
tables = client.discover("snowflake", secrets=secrets)
result = client.process_from_connector("snowflake", table="ORDERS",
    target_columns=["AMOUNT"], secrets=secrets)

BigQuery:

secrets = {
    "credentials_json": '{"type":"service_account",...}',
    "project_id": "my-project",
    "dataset": "analytics",
}
tables = client.discover("bigquery", secrets=secrets)

Amazon S3:

secrets = {
    "bucket": "my-data-lake",
    "access_key_id": "AKIA...",
    "secret_access_key": "...",
    "region": "us-east-1",
}
result = client.process_from_connector("s3", table="data/sales.csv",
    target_columns=["revenue"], secrets=secrets, output_rows=50000)

MongoDB:

secrets = {
    "connection_string": "mongodb+srv://user:[e-mail]",
    "database": "analytics",
}
tables = client.discover("mongodb", secrets=secrets)

---

client.test_connector(connector_type, secrets, credential_id) — Test connection

result = client.test_connector("s3", secrets={...})
# {"success": True, "message": "Connection successful"}

client.discover(connector_type, ...) — List tables / files

tables = client.discover("sql", secrets={...})
# ["users", "orders", "transactions"]

client.preview(connector_type, table, ...) — Preview columns & row count

info = client.preview("sql", table="orders", secrets={...})
# {"success": True, "columns": ["id", "price", ...], "row_count": 50000}

Supports optional custom_query parameter for SQL-based connectors:

info = client.preview("sql", table="orders", custom_query="SELECT * FROM orders WHERE date > '2025-01-01'", secrets={...})

client.process_from_connector(...) — Run pipeline from connector

result = client.process_from_connector(
    connector_type="sql",
    table="orders",
    target_columns=["price"],
    secrets={"connection_string": "postgresql://..."},
    output_rows=20000,
    incremental=True,          # Only fetch new rows since last sync
    type_casts={"age": "int"}, # Override column types before processing
    wait=True,
    download_path="./outputs/",
)

Supports all the same options as client.process(): tools, advanced_params, wait, poll_interval, auto_download, output_preferences, compressed.

client.write_output(session_id, ...) — Write results to a target

client.write_output(
    session_id="abc123...",
    connector_type="sql",
    table_name="ml_prepared_data",
    secrets={"connection_string": "postgresql://..."},
    output_stage="dsg",        # which pipeline output to write
    if_exists="replace",       # "replace", "append", "fail", or "merge"
    merge_keys=["id"],         # column names for merge matching (when if_exists='merge')
)
# {"success": True, "rows_written": 20000, "table": "ml_prepared_data"}

---

Credential Management

Save credentials on the server so you don't need to pass secrets every time:

# Save a new credential
cred = client.save_credential(
    name="Production DB",
    connector_type="sql",
    secrets={"connection_string": "postgresql://..."},
)
cred_id = cred["id"]

# Use credential_id instead of secrets
tables = client.discover("sql", credential_id=cred_id)
result = client.process_from_connector("sql", table="orders",
    target_columns=["price"], credential_id=cred_id)

# List saved credentials (secrets are never exposed)
creds = client.list_credentials()

# Update a credential
client.update_credential(cred_id, name="Prod DB v2", secrets={...})

# Delete a credential
client.delete_credential(cred_id)

---

Schema Mapping

Auto-suggest and apply column mappings between source and target schemas:

client.suggest_mapping(source_columns, target_columns) — Auto-suggest mapping

mapping = client.suggest_mapping(
    source_columns=["customer_name", "order_amt", "order_dt"],
    target_columns=["name", "amount", "date"],
    threshold=0.6,  # minimum similarity score (0-1)
)
# {
#   "mapping": {"customer_name": "name", "order_amt": "amount", "order_dt": "date"},
#   "scores": {"customer_name": 0.85, "order_amt": 0.72, "order_dt": 0.68}
# }

client.apply_mapping(session_id, mapping) — Rename columns

result = client.apply_mapping(
    session_id="abc123...",
    mapping={"old_col_name": "new_col_name", ...},
    output_stage="dsg",  # which pipeline stage to apply to
)
# {"columns": ["new_col_name", ...], "preview": [...]}

---

Quality Alerts

Monitor pipeline metrics and get notified when thresholds are breached:

# Create a quality alert rule
rule = client.create_quality_alert(
    name="High row loss",
    metric="row_loss_pct",    # row_loss_pct, null_pct, column_drop_count, duration_seconds
    operator=">",             # >, <, >=, <=, ==
    threshold=10.0,
    severity="critical",      # warning or critical
    stage="mdh",              # optional: anomaly, dtc, mdh, cds, dsm, dsg
)

# List rules
rules = client.list_quality_alerts()

# Update a rule
client.update_quality_alert(rule_id="...", threshold=15.0, severity="warning")

# Get fired alert events
events = client.get_alert_events(session_id="optional-filter")

# Delete a rule
client.delete_quality_alert(rule_id="...")

Available metrics:

Metric	Description
row_loss_pct	Percentage of rows lost in a step
null_pct	Percentage of null values remaining
column_drop_count	Number of columns dropped
duration_seconds	Time taken by a pipeline step

---

Sync Watermarks

Track incremental sync progress for connector-based pipelines:

# List all watermarks
watermarks = client.list_watermarks()
# [{"id": "wm-123", "connector_type": "sql", "table_name": "orders",
#   "watermark_column": "updated_at", "last_value": "2026-04-12T00:00:00Z"}]

# Reset a watermark (re-sync from beginning)
client.reset_watermark(watermark_id="wm-123")

---

Scheduled Runs

Automate recurring pipeline executions:

# Create an interval-based schedule (every 24 hours)
schedule = client.create_scheduled_run(
    name="Daily ETL",
    schedule_type="interval",
    interval_minutes=1440,
    connector_type="snowflake",
    table="orders",
    credential_id="cred-id",
    y_columns=["target"],
    output_rows=50000,
)

# Create with cron expression (weekdays at 8 AM)
schedule = client.create_scheduled_run(
    name="Weekday ETL",
    schedule_type="cron",
    cron_expression="0 8 * * 1-5",
    connector_type="snowflake",
    table="orders",
    credential_id="cred-id",
    y_columns=["target"],
    tools={"anomaly": True, "dsg": False},  # customize pipeline steps
)

# List all schedules
schedules = client.list_scheduled_runs()

# Enable/disable a schedule
client.toggle_scheduled_run(run_id="run-id")

# Trigger immediately (ignore schedule)
client.run_scheduled_now(run_id="run-id")

# Delete a schedule
client.delete_scheduled_run(run_id="run-id")

---

Folder Listeners

Automatically trigger pipelines when new files appear in cloud storage:

# Create an S3 folder listener
listener = client.create_listener(
    name="S3 Ingest",
    source_type="s3",            # s3, gcs, azure_blob, local, sftp
    watch_path="s3://bucket/incoming/",
    credential_id="cred-id",
    y_columns=["target"],
    pipeline_config={"enable_dsg": False},
    output_rows=10000,
)

# List all listeners
listeners = client.list_listeners()

# Update a listener
client.update_listener(listener_id="...", watch_path="s3://bucket/new-path/", enabled=False)

# Delete a listener
client.delete_listener(listener_id="...")

Supported source types: s3, gcs, azure_blob, local, sftp

---

SFTP Upload

Upload files to the AutoData server via SFTP for processing:

# Get SFTP server connection info
info = client.sftp_info()
# {"host": "...", "port": 22, "username": "..."}

# Manage SFTP credentials
cred = client.create_sftp_credential(name="My Upload Key")
# {"id": "...", "username": "...", "password": "..."} (password shown only once)

creds = client.list_sftp_credentials()
client.delete_sftp_credential(credential_id="...")

---

Spark (Large Data)

For datasets too large for pandas, AutoData can use PySpark on the server:

client.spark_status() — Check Spark availability

status = client.spark_status()
# {"available": True, "spark_version": "3.5.0", ...}

client.spark_read(file_path, sample_rows) — Read large files

info = client.spark_read(
    file_path="/data/huge_dataset.parquet",
    sample_rows=20,
)
# {"row_count": 50000000, "columns": [...], "dtypes": {...}, "sample": [...]}

client.spark_transform(...) — Apply Spark transformations

result = client.spark_transform(
    file_path="/data/huge_dataset.parquet",
    output_path="/data/transformed.csv",
    operations=[
        {"op": "cast", "type_casts": {"age": "int"}},
        {"op": "drop_nulls", "subset": ["col1"]},
        {"op": "fill_nulls", "strategy": "mean"},
        {"op": "normalize", "method": "minmax"},
        {"op": "sample", "n": 50000},
    ],
    output_format="csv",  # csv or parquet
)
# {"row_count": 50000, "columns": [...]}

---

Worker Status

Check the processing worker fleet:

status = client.worker_status()
# {"backend": "local", "active_jobs": 2, "queue_size": 0, ...}

---

Error Handling

All API errors raise AutoDataError:

from datatoolpack import AutoDataClient, AutoDataError

with AutoDataClient() as client:
    try:
        result = client.process("data.csv", target_columns="price")
    except AutoDataError as e:
        print(f"API error {e.status_code}: {e}")
    except FileNotFoundError as e:
        print(f"File not found: {e}")
    except ValueError as e:
        print(f"Invalid input: {e}")  # e.g. unsupported file format

AutoDataError attributes:

• str(e) — human-readable error message from the server
• e.status_code — HTTP status code (e.g. 401, 429, 500), or None for non-HTTP errors

Transient errors (429, 502, 503, 504) are automatically retried up to max_retries times with exponential back-off.

---

Advanced Example: Non-blocking with manual polling

import time
from datatoolpack import AutoDataClient, AutoDataError

with AutoDataClient() as client:
    # Start job without blocking
    job = client.process(
        "large_dataset.csv",
        target_columns=["churn"],
        wait=False,
    )
    session_id = job["session_id"]
    print(f"Job started: {session_id}")

    # Poll manually
    while True:
        status = client.get_status(session_id)
        print(f"  {status['progress_percent']}% — {status['message']}")
        if status["status"] == "completed":
            break
        elif status["status"] in ("error", "cancelled"):
            raise AutoDataError(f"Job {status['status']}: {status['message']}")
        time.sleep(5)

    # Download results
    path = client.download_results(session_id, download_path="./outputs/")
    print(f"Results saved to {path}")

---

Complete Method Reference

Core Pipeline

Method	Description
process(file_path, target_columns, ...)	Upload a file and run the pipeline
get_status(session_id)	Poll processing progress
get_result(session_id)	Get completed session results
wait_for_completion(session_id)	Block until session finishes
cancel(session_id)	Cancel a running job
retry_session(session_id)	Retry a failed session from checkpoint
download_results(session_id, ...)	Download result files
download_file(url, output_path)	Download a single file

API Keys & Usage

Method	Description
list_keys()	List all API keys for the account
get_usage()	Get credit usage statistics

Connectors

Method	Description
test_connector(connector_type, ...)	Test connectivity to a data source
discover(connector_type, ...)	List tables/files in a data source
preview(connector_type, table, ...)	Preview columns and row count
process_from_connector(connector_type, table, ...)	Run pipeline from a connector source
write_output(session_id, connector_type, ...)	Write results to a database/storage target

Credentials

Method	Description
list_credentials()	List saved credentials
save_credential(name, connector_type, secrets)	Save a new credential
update_credential(credential_id, ...)	Update a saved credential
delete_credential(credential_id)	Delete a saved credential

Schema Mapping

Method	Description
suggest_mapping(source_columns, target_columns)	Auto-suggest column mapping
apply_mapping(session_id, mapping)	Rename columns using a mapping

Quality Alerts

Method	Description
list_quality_alerts()	List quality alert rules
create_quality_alert(name, metric, operator, threshold, ...)	Create an alert rule
update_quality_alert(rule_id, ...)	Update an alert rule
delete_quality_alert(rule_id)	Delete an alert rule
get_alert_events(session_id=None)	List fired alert events

Sync Watermarks

Method	Description
list_watermarks()	List all sync watermarks
reset_watermark(watermark_id)	Reset a watermark to re-sync

Scheduled Runs

Method	Description
list_scheduled_runs()	List all scheduled runs
create_scheduled_run(name, schedule_type, ...)	Create a scheduled run
delete_scheduled_run(run_id)	Delete a scheduled run
toggle_scheduled_run(run_id)	Enable/disable a scheduled run
run_scheduled_now(run_id)	Trigger a scheduled run immediately

Folder Listeners

Method	Description
list_listeners()	List folder listeners
create_listener(name, source_type, watch_path, ...)	Create a folder listener
update_listener(listener_id, ...)	Update a folder listener
delete_listener(listener_id)	Delete a folder listener

SFTP

Method	Description
sftp_info()	Get SFTP server connection info
list_sftp_credentials()	List SFTP credentials
create_sftp_credential(name)	Create SFTP credential
delete_sftp_credential(credential_id)	Delete SFTP credential

Spark (Large Data)

Method	Description
spark_status()	Check if Spark is available
spark_read(file_path, sample_rows)	Read a large file via Spark
spark_transform(file_path, output_path, operations, ...)	Apply Spark transformations

Worker Status

Method	Description
worker_status()	Get worker fleet status

---

Requirements

• Python >= 3.8
• requests >= 2.25.0

License

MIT