lht: Lakehouse Tools

Bring Your Own Data Warehouse.

lht is an open-source Python library and CLI that moves Salesforce data into your warehouse and pushes it back. It uses the Salesforce Bulk API 2.0 and runs on your own infrastructure, with your own credentials. There is no hosted service and no per-row pricing, and your data doesn't pass through a third party.

pip install "lht[snowflake]"
lht sync --sobject Account --table ACCOUNT

Supported today: Salesforce ↔ Snowflake, and Salesforce ← Databricks (reverse ETL and merge, in a notebook or a Job — see Databricks). The connection layer is built so more warehouses can be added; see Roadmap.

---

What it does

Salesforce → Snowflake sync	Full load on the first run. After that, incremental loads use LastModifiedDate and a MERGE on the record Id. Tables are created for you, with Salesforce field types mapped to Snowflake types.
Reverse ETL (warehouse → Salesforce)	upsert, insert, update and delete records in Salesforce from any Snowflake or Databricks SELECT. Runs through Bulk API 2.0 ingest jobs, with optional result logging back into the warehouse.
Databricks reverse ETL	Databricks ingests Salesforce well and has no way to write back. lht runs on your Databricks compute, reads Unity Catalog with Spark SQL, and pushes to Salesforce — from a notebook, a Job, or locally via Databricks Connect.
Record merge	Deduplicate Accounts, Contacts or Leads inside Salesforce from a warehouse query that returns MasterId / LoserId pairs. Uses the SOAP merge() call, which the Bulk API can't do. --dry-run validates the pairs first.
Bulk job management	List, inspect and delete Bulk API 2.0 jobs, and download their success, failure and unprocessed results.
Headless auth	Supports the OAuth 2.0 Client Credentials flow and JWT bearer flow for Salesforce, and key-pair (JWT) auth for Snowflake. Credentials can come from a local config file or from your own secret store.

Why lht

• You own the pipeline. It's a pip install, not a SaaS contract. Run it from a laptop, cron, Airflow, Dagster, GitHub Actions, or a Snowflake/Databricks job.
• The warehouse is a choice, not a rewrite. The same reverse-ETL call takes a Snowpark session or a Spark session; the SQL and the Salesforce side don't change.
• Bulk API 2.0 everywhere. Reads and writes both use Bulk API 2.0, so large objects don't burn through your REST API call limits.
• Two directions, one tool. Sync, reverse ETL and merge share the same saved connections.
• Plain Python, Apache 2.0. Read it, fork it, extend it.

Installation

From PyPI (recommended)

Requires Python 3.9 or newer.

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install "lht[snowflake]"       # or "lht[databricks]", or both: "lht[all]"
lht --help

The warehouse client is an extra, so pick the one you use:

Install	For
pip install "lht[snowflake]"	Snowflake — sync, reverse ETL, merge
pip install "lht[databricks]"	Databricks from a laptop or CI, via Databricks Connect. Needs Python 3.10 or newer: the Databricks Connect releases it installs don't support 3.9
pip install lht	On Databricks compute, where spark already exists. Nothing is installed over the runtime's pandas/pyarrow
pip install "lht[all]"	Both warehouses

To upgrade: pip install --upgrade "lht[snowflake]". To pin a version: pip install "lht[snowflake]==3.0.0".

From source

git clone https://github.com/lakehousetools/lht.git
cd lht
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"            # editable install, plus pytest/build/twine
pytest                             # unit tests; no Salesforce or Snowflake needed

You can also download a release archive from GitHub Releases or the source distribution from PyPI and run pip install lht-<version>.tar.gz.

Quickstart

1. Prepare Salesforce

Create an External Client App (or Connected App) with the OAuth 2.0 Client Credentials flow enabled and a run-as user assigned. Note its Consumer Key, Consumer Secret and your My Domain (the part before .my.salesforce.com). See docs/authentication.md for step-by-step setup, including the JWT bearer flow.

Start with a sandbox or Developer Edition org. Reverse ETL and merge change data in Salesforce.

2. Prepare Snowflake

Create a user with key-pair authentication and a role that can use a warehouse and create tables in your target schema:

CREATE ROLE IF NOT EXISTS LHT_ROLE;
GRANT USAGE ON WAREHOUSE COMPUTE_WH TO ROLE LHT_ROLE;
GRANT USAGE ON DATABASE SALESFORCE TO ROLE LHT_ROLE;
GRANT USAGE, CREATE TABLE ON SCHEMA SALESFORCE.RAW TO ROLE LHT_ROLE;
GRANT ROLE LHT_ROLE TO USER LHT_USER;
ALTER USER LHT_USER SET RSA_PUBLIC_KEY = '<contents of rsa_key.pub>';

3. Save connections

lht create-connection --snowflake     # account, user, role, warehouse, private key, database, schema
lht create-connection --salesforce    # client id, client secret, My Domain, sandbox y/n
lht list-connections
lht connect my_salesforce             # verify it works

Connections are stored in connections.toml under lht's config directory (~/.lakehousetools by default — see Authentication for the full resolution order) with owner-only permissions (0600). Snowflake private keys are copied into the same directory.

4. Sync

lht sync --sobject Account --table ACCOUNT --schema RAW
lht sync --sobject Contact --table CONTACT --schema RAW --where "IsDeleted = false"

Run the same command again to pick up only records changed since the last sync.

5. Push data back (reverse ETL)

lht retl upsert --sobject Account --match-field External_Id__c \
  --sql "SELECT External_Id__c, Name, Industry FROM ANALYTICS.ACCOUNT_ENRICHED"

Column names in the query must match Salesforce field API names.

Python API

from lht.user.auth import create_session
from lht.user.salesforce_auth import get_salesforce_access_info
from lht.salesforce.intelligent_sync import sync_sobject_intelligent
from lht.salesforce import retl

session = create_session(connection_name="my_snowflake")           # Snowpark session
access_info = get_salesforce_access_info("my_salesforce")          # {'access_token', 'instance_url'}

result = sync_sobject_intelligent(
    session=session,
    access_info=access_info,
    sobject="Opportunity",
    schema="RAW",
    table="OPPORTUNITY",
)
print(result["sync_method"], result["actual_records"])

retl.upsert(
    session, access_info,
    sobject="Account",
    query="SELECT External_Id__c, Rating FROM ANALYTICS.ACCOUNT_SCORES",
    field="External_Id__c",
)

To run without a local config file (CI, Airflow, Snowflake tasks), pass credentials from your secret store directly:

from lht.user.salesforce_auth import get_salesforce_access_info_from_credentials

access_info = get_salesforce_access_info_from_credentials({
    "auth_flow": "jwt_bearer",
    "client_id": os.environ["SF_CLIENT_ID"],
    "username": "integration@example.com",
    "private_key_pem": os.environ["SF_PRIVATE_KEY"],
    "sandbox": True,
})

More in docs/python-api.md and examples/.

CLI reference

Command	Purpose
lht create-connection --snowflake | --salesforce	Save a connection interactively
lht list-connections / edit-connection / set-primary NAME / connect NAME	Manage and test connections
lht sync --sobject OBJ --table TABLE [--schema --database --where --force-full-sync --use-stage --stage-name]	Salesforce → Snowflake
lht retl {upsert,insert,update,delete} --sobject OBJ (--sql SQL | --sql-file FILE) [--match-field --batch-size --clear-nulls --log-results]	Warehouse → Salesforce
lht merge --sobject {Account,Contact,Lead} (--sql | --sql-file) [--dry-run]	Merge duplicate records in Salesforce
lht list-jobs / show-job ID / get-job-results ID / delete-job ID	Bulk API 2.0 job management

Every command accepts --snowflake NAME / --salesforce NAME to override the primary connection. retl and merge also accept --databricks [PROFILE] in place of --snowflake, to read from Databricks instead. Full details are in docs/cli.md, or run lht <command> --help.

How incremental sync works

1. If the target table doesn't exist, lht creates it from the SObject describe and runs a full Bulk API 2.0 query.
2. If it exists, lht reads MAX(LASTMODIFIEDDATE) from the table and queries only newer records.
3. Result batches load into a temporary table and are MERGEd into the target on ID.
4. The Bulk API job is deleted afterwards unless you pass --no-delete-job.

Queries use queryAll, so records deleted in Salesforce arrive with IsDeleted = true while they're in the Recycle Bin. Records that are hard-deleted or purged before the next sync aren't seen; run --force-full-sync periodically if that matters to you.

How lht compares

	lht	Managed ELT services	Hand-written scripts
Where it runs	Your infrastructure	Vendor cloud	Your infrastructure
Cost model	Free (Apache 2.0) + your warehouse compute	Subscription / usage-based	Engineering time
Salesforce → warehouse	✅ Bulk API 2.0, incremental	✅	You build it
Warehouse → Salesforce (reverse ETL)	✅ Built in	Often a separate product	You build it
Record merge / dedupe in Salesforce	✅ lht merge	Rare	You build it
Data leaves your control	No	Yes, transits the vendor	No

lht is a good fit if you already run Snowflake, want Salesforce data there on your own schedule, and want reverse ETL without adding another vendor.

Security

• Credentials are stored locally in lht's config directory (~/.lakehousetools by default; directory 0700, files 0600), or passed in from your own secret store, or registered in memory by name with register_connection().
• Salesforce secrets are sent in POST bodies, never URLs, and are never logged.
• Every HTTP call has a timeout, and database, schema and table names are validated before they go into SQL.

To report a vulnerability, see SECURITY.md.

Databricks

Databricks ingests Salesforce well, and has no reverse ETL. In a notebook:

%pip install lht
dbutils.library.restartPython()

from lht.salesforce import retl
from lht.warehouse.databricks import connect, salesforce_access_info

retl.upsert(
    connect(),                                  # the cluster's `spark`
    salesforce_access_info(scope="salesforce"),  # a Databricks secret scope
    sobject="Account",
    query="SELECT external_id AS Ext__c, rating AS Rating FROM main.analytics.account_scores",
    field="Ext__c",
)

The same file runs on a laptop against a ~/.databrickscfg profile (pip install "lht[databricks]"), or from the CLI with lht retl upsert --databricks my-workspace .... See Databricks and examples/databricks_retl.py.

Documentation

• Getting started
• Authentication (Salesforce & Snowflake)
• CLI reference
• Python API
• Reverse ETL
• Databricks
• Merging records
• Intelligent sync internals
• FAQ

Getting help

I want to…	Go to
Ask a question or get setup help	Discussions → Q&A
Share an idea or how you use lht	Discussions → Ideas / Show and tell
Report a bug or request a feature	Issues
Report a security vulnerability	Private report (see SECURITY.md)

Please remove credentials, org names and customer data from anything you post.

Roadmap

• Salesforce → Databricks sync (reverse ETL and merge already work there; on Databricks, Lakeflow Connect handles ingestion today)
• Additional warehouses (BigQuery, Postgres) behind the same CLI
• Hard-delete detection for incremental sync
• Scheduling recipes (Airflow, Dagster, Snowflake Tasks)

Ideas are welcome in Discussions; pull requests are welcome too.

Contributing

See CONTRIBUTING.md for development setup, tests and the release process. By participating you agree to the Code of Conduct.

License

Apache License 2.0