Metadata-Version: 2.5
Name: dbhop
Version: 0.1.0
Summary: Move data between SQL databases, Parquet files and S3: chunked exports, typed imports with auto-generated DDL, resumable upserts, and built-in progress tracking.
Project-URL: Homepage, https://github.com/arirismansyah/dbhop
Project-URL: Documentation, https://github.com/arirismansyah/dbhop#readme
Project-URL: Issues, https://github.com/arirismansyah/dbhop/issues
Project-URL: Changelog, https://github.com/arirismansyah/dbhop/blob/main/CHANGELOG.md
Author: Ari Rismansyah
License-Expression: MIT
License-File: LICENSE
Keywords: data-export,data-import,ddl,duckdb,etl,minio,mysql,parquet,postgresql,s3,sql-server,upsert
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Scientific/Engineering :: Information Analysis
Classifier: Topic :: Utilities
Requires-Python: >=3.10
Requires-Dist: boto3>=1.28
Requires-Dist: duckdb>=1.0
Requires-Dist: numpy
Requires-Dist: pandas>=2.0
Requires-Dist: pyarrow>=14
Requires-Dist: python-dotenv>=1.0
Requires-Dist: sqlalchemy>=2.0
Requires-Dist: tqdm>=4.60
Provides-Extra: all
Requires-Dist: openpyxl>=3.1; extra == 'all'
Requires-Dist: psycopg2-binary>=2.9; extra == 'all'
Requires-Dist: pymysql>=1.1; extra == 'all'
Requires-Dist: pyodbc>=5.0; extra == 'all'
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: twine>=6.1; extra == 'dev'
Provides-Extra: excel
Requires-Dist: openpyxl>=3.1; extra == 'excel'
Provides-Extra: mssql
Requires-Dist: pyodbc>=5.0; extra == 'mssql'
Provides-Extra: mysql
Requires-Dist: pymysql>=1.1; extra == 'mysql'
Provides-Extra: notebook
Requires-Dist: ipywidgets; extra == 'notebook'
Provides-Extra: postgres
Requires-Dist: psycopg2-binary>=2.9; extra == 'postgres'
Description-Content-Type: text/markdown

# dbhop

Move data between **SQL databases**, **Parquet files**, and **S3**, with one Python API and one CLI.

- **Export** tables or query results from PostgreSQL / MySQL / SQL Server to **Parquet**, streamed in chunks, so tables with hundreds of millions of rows never blow up RAM. Then upload them to **AWS S3 or any S3-compatible storage** (MinIO, Ceph, Cloudflare R2, ...).
- **Import** CSV / Excel / Parquet / JSON files from **local disk or S3** into a database. `dbhop` generates the `CREATE TABLE` from the actual data, creates the table, and bulk-loads it with the fastest path per database (`COPY`, `fast_executemany`, multi-row `INSERT`). Every step can also be used on its own, e.g. *just generate the DDL*.
- **Upsert / delete** Parquet data into SQL Server with chunked `MERGE`, row-level locks, and optional parallel workers.
- **Resumable**: imports, upserts, and deletes checkpoint every batch. Run the same command again after a crash and it picks up where it stopped, with no duplicate rows.
- **Observable**: progress bars with ETA in terminals and Jupyter, periodic log lines under cron/nohup, and a `dbhop status` command to watch background jobs from another terminal.
- **Portable**: the same code runs on a laptop and on a server. Database drivers (ODBC 17/18, FreeTDS, psycopg2/psycopg 3) are **detected at runtime**.

```python
from dbhop import DatabaseClient, S3Client, export_table_to_s3, import_file

db = DatabaseClient("postgresql://user:pass@localhost:5432/shop")

# database -> parquet -> s3://my-bucket/backup/orders/dt=2026-10-20/orders.parquet
export_table_to_s3(db, S3Client(), "orders", dest="s3://my-bucket/backup/{table}/dt={date}/")

# CSV -> CREATE TABLE (types & lengths measured from the data) -> bulk insert
import_file("customers.csv", db, schema="staging", normalize_columns=True)
```

---

## Contents

1. [Installation](#installation)
2. [Configuration](#configuration)
3. [Checking your setup: `dbhop doctor`](#checking-your-setup-dbhop-doctor)
4. [Guides & examples](#guides--examples)
5. [Exporting to S3](#exporting-to-s3)
6. [Importing files into a database](#importing-files-into-a-database)
7. [Upsert & delete into SQL Server](#upsert--delete-into-sql-server)
8. [Progress & monitoring](#progress--monitoring)
9. [CLI](#cli)
10. [API reference](#api-reference)
11. [Driver detection (laptop vs. server)](#driver-detection-laptop-vs-server)
12. [Troubleshooting](#troubleshooting)
13. [Development](#development)
14. [License](#license)

---

## Installation

Requires Python 3.10+.

```bash
pip install "dbhop[postgres]"      # pick the extras you need
```

| Extra | Installs | Needed for |
| --- | --- | --- |
| `postgres` | `psycopg2-binary` | PostgreSQL (psycopg 3 also works and is detected automatically) |
| `mysql` | `pymysql` | MySQL / MariaDB (pure Python, nothing to compile) |
| `mssql` | `pyodbc` | SQL Server (also needs an OS-level ODBC driver, see below) |
| `excel` | `openpyxl` | Reading `.xlsx` files (add `xlrd` for legacy `.xls`) |
| `notebook` | `ipywidgets` | Widget-style progress bars in Jupyter |
| `all` | `postgres` + `mysql` + `mssql` + `excel` | Everything |

The package is installed and imported as `dbhop` and provides the `dbhop` command (`python -m dbhop` works too). Database drivers are imported lazily, so a machine that only talks to Postgres never needs ODBC.

### ODBC driver (SQL Server only)

`pyodbc` needs Microsoft's ODBC driver installed on the OS. Install **one** version (18 recommended; 17 works too). If a machine already has Driver 17 or FreeTDS, there's no need to reinstall; dbhop uses what's there.

#### macOS (Homebrew)

```bash
brew tap microsoft/mssql-release https://github.com/Microsoft/homebrew-mssql-release
HOMEBREW_ACCEPT_EULA=Y brew install msodbcsql18
```

#### Ubuntu / Debian

```bash
curl -fsSL https://packages.microsoft.com/keys/microsoft.asc | sudo gpg --dearmor -o /usr/share/keyrings/microsoft-prod.gpg
curl -fsSL https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/prod.list | sudo tee /etc/apt/sources.list.d/mssql-release.list
sudo apt-get update
sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18 unixodbc-dev
```

#### RHEL / Rocky / CentOS

```bash
curl -fsSL https://packages.microsoft.com/config/rhel/9/prod.repo | sudo tee /etc/yum.repos.d/mssql-release.repo
sudo ACCEPT_EULA=Y yum install -y msodbcsql18 unixODBC-devel
```

**Windows**: download the "ODBC Driver 18 for SQL Server" installer from Microsoft.

---

## Configuration

Connections can be passed directly in code (`DatabaseClient("postgresql://...")`), but the usual approach is environment variables, typically from a `.env` file. Copy [`.env.example`](.env.example) to `.env`:

```dotenv
# database connection: <PREFIX>_TYPE/HOST/PORT/USER/PASSWORD/NAME
DB_TYPE=mssql
DB_HOST=10.0.0.5
DB_PORT=1433
DB_USER=etl_user
DB_PASSWORD=p@ss$word#safe
DB_NAME=warehouse

# S3: all defaults, the destination can be changed per call
AWS_ACCESS_KEY=...
AWS_SECRET_KEY=...
# leave empty for AWS; set it for MinIO / Ceph / R2 / etc.
S3_ENDPOINT=https://s3.example.com
S3_BUCKET=default-bucket
```

- **Several databases**: use different prefixes, e.g. `WAREHOUSE_DB_HOST`, `CRM_DB_HOST`, then `DatabaseClient.from_env("WAREHOUSE_DB")` / `dbhop ... --db-env WAREHOUSE_DB`.
- Passwords with special characters (`@ $ # ; }`) can be written as-is, without encoding.
- Put comments on **their own line**. A comment after an empty value (`KEY=   # ...`) is read as the value.
- The standard AWS names are recognized too: `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_DEFAULT_REGION`. With no keys at all, boto3's default credential chain is used (`~/.aws/credentials`, IAM roles, ...).
- The CLI reads `.env` from the current directory automatically (`--env-file path/.env` for another file, placed **before** the command name). In Python, call `dotenv.load_dotenv()` yourself.
- Never commit `.env`.

---

## Checking your setup: `dbhop doctor`

Run this **first on every machine** (laptop, server, a colleague's PC):

```bash
dbhop doctor --db-env DB --check-s3
```

```text
Postgres driver:
  in use    : psycopg2

ODBC driver (for mssql):
  installed : ['FreeTDS', 'ODBC Driver 18 for SQL Server']
  in use    : ODBC Driver 18 for SQL Server

Database connection test:
  mssql -> 10.0.0.5:1433/warehouse
  [ok] connected

S3 connection test:
  endpoint https://s3.example.com | bucket default-bucket
  [ok] bucket is accessible

Result: READY
```

Exit code `0` means ready; `1` means something is wrong, with the explanation printed above.

---

## Guides & examples

| Guide | Contents |
| --- | --- |
| [Export tables to S3](docs/export-table-to-s3.md) | Step by step: prepare bucket & folder → explore database/schema/tables → export → monitor → verify. Includes flowcharts. |
| [Notebook: export tables to S3](examples/notebooks/export_table_to_s3.ipynb) | The guide above as a notebook: fill in the settings and run the cells in order |
| [Import local files into a table](docs/import-file-to-table.md) | Step by step: inspect the file → new vs. existing table → review DDL → import → verify. Includes resuming interrupted imports. |
| [Notebook: import a local file](examples/notebooks/import_file_to_table.ipynb) | The guide above as a notebook, using the sample file `examples/data/sample_customers.csv` |

Ready-to-adapt scripts in [`examples/`](examples/):

| File | Contents |
| --- | --- |
| [01_export_table_to_s3.py](examples/01_export_table_to_s3.py) | Export a full table, filter columns/rows, custom file name, keep a local copy |
| [02_export_query_to_s3.py](examples/02_export_query_to_s3.py) | Export a query result, bind params, `IN` clauses, queries from `.sql` files |
| [03_export_many_tables.py](examples/03_export_many_tables.py) | Back up many tables at once, safe to re-run |
| [04_s3_utilities.py](examples/04_s3_utilities.py) | Upload files/folders, list, download, presigned URLs |
| [05_s3_to_upsert_mssql.py](examples/05_s3_to_upsert_mssql.py) | Fetch parquet from S3, then upsert into SQL Server |
| [06_generate_ddl.py](examples/06_generate_ddl.py) | Generate DDL only from local/S3 files, and edit the column plan |
| [07_import_file_to_db.py](examples/07_import_file_to_db.py) | One-step imports: local, S3, folders, Excel, one file = one table |
| [08_insert_existing_table.py](examples/08_insert_existing_table.py) | Insert into an existing table, rename columns, truncate |
| [09_monitor_progress.py](examples/09_monitor_progress.py) | Read progress/ETA of running jobs & the history of failed jobs from Python |

---

## Exporting to S3

```python
from dotenv import load_dotenv
from dbhop import DatabaseClient, S3Client, export_table_to_s3

load_dotenv()
db = DatabaseClient.from_env("DB")
s3 = S3Client()

export_table_to_s3(
    db, s3, "orders", schema="dbo",
    dest="s3://my-bucket/backup/{table}/dt={date}/",
)
# -> s3://my-bucket/backup/orders/dt=2026-10-20/orders.parquet
```

Behind the scenes:

1. The table is read in chunks (default 100,000 rows) and written to a temporary Parquet file.
2. The file is uploaded (multipart & parallel for large files).
3. The S3 object's size is checked against the local file.
4. The temporary file is deleted.

The function returns:

```python
{'status': 'uploaded', 's3_uri': 's3://my-bucket/backup/...', 'bucket': 'my-bucket',
 'key': 'backup/...', 'rows': 1250000, 'size_bytes': 48211234, 'duration_sec': 42.7,
 'local_path': None}
```

### Choosing the destination

The destination is **always** chosen by the caller. `S3_BUCKET` is only used when no bucket is given.

| Form | Example | Resulting key |
| --- | --- | --- |
| `dest` as a folder (ends with `/`) | `dest="s3://lake/backup/"` | `lake` → `backup/<table>.parquet` |
| `dest` as a file | `dest="s3://lake/backup/orders.parquet"` | `lake` → `backup/orders.parquet` |
| `bucket` + `prefix` | `bucket="lake", prefix="a/b"` | `lake` → `a/b/<table>.parquet` |
| + `filename` | `bucket="lake", prefix="a", filename="x.parquet"` | `lake` → `a/x.parquet` |

**Placeholders** available in `dest`, `prefix`, and `filename`:

| Placeholder | Example value |
| --- | --- |
| `{table}` | `orders` (for queries: the `name` value) |
| `{schema}` | `dbo` |
| `{date}` | `2026-10-20` |
| `{timestamp}` | `20261020_153000` |
| `{year}` `{month}` `{day}` | `2026` `10` `20` |

Common patterns:

```python
dest="s3://lake/raw/{table}/dt={date}/"              # daily partitions
prefix="snapshot/{year}/{month}", filename="{table}_{timestamp}.parquet"
```

To target another storage/endpoint, create another client:

```python
s3_minio = S3Client(endpoint_url="http://10.0.0.9:9000", access_key="...", secret_key="...")
```

---

## Importing files into a database

> Step-by-step guide for local files: [docs/import-file-to-table.md](docs/import-file-to-table.md) (notebook version: [import_file_to_table.ipynb](examples/notebooks/import_file_to_table.ipynb)).

### Flow

```text
 source                      read (DuckDB)              DDL                       insert
 ─────────────────────       ──────────────────         ───────────────────       ──────────────────────
 file / glob / folder   ──►  column types inferred ──►  CREATE TABLE fitted  ──►  per batch + checkpoint
 s3://... (downloaded)       + text lengths measured    to the data (or use        pg: COPY
 pandas DataFrame                                       an existing table)         mssql: fast_executemany
                                                                                   mysql: executemany
```

Each step can be run on its own:

| Need | Function | CLI |
| --- | --- | --- |
| See a file's contents & types | `FileSource(src).print_info()` | `dbhop inspect <src>` |
| DDL only (no database) | `generate_ddl(src, "mssql", ...)` | `dbhop ddl <src> --db-type mssql` |
| Insert into an existing table | `insert_file(src, db, "table")` | `dbhop insert <src> --table table` |
| Everything at once | `import_file(src, db, ...)` | `dbhop import <src>` |
| Many files, 1 file = 1 table | `import_files(src, db, ...)` | `dbhop import <folder> --each` |

### Supported sources

| `source` | Meaning |
| --- | --- |
| `"data/customers.csv"` | one local file |
| `"data/*.csv"`, `"data/folder/"` | many local files, **combined into one table** |
| `"s3://bucket/raw/customers.parquet"` | one S3 object |
| `"s3://bucket/raw/orders/"`, `"s3://bucket/raw/part-*.parquet"` | many S3 objects, combined |
| `pandas.DataFrame` | in-memory data (`table_name` required) |
| `["a.csv", "s3://b/c.csv"]` | a mixed list (must share a format) |

Formats: `.parquet`, `.csv`, `.tsv`, `.txt`, `.json`, `.jsonl`/`.ndjson`, `.xlsx`/`.xls`, including compressed variants like `.csv.gz`. S3 files are first downloaded to a temporary folder and deleted afterwards, so custom endpoints work too.

### Examples

```python
from dbhop import DatabaseClient, import_file, insert_file, generate_ddl

db = DatabaseClient.from_env("WAREHOUSE_DB")

# one step: S3 -> new table
import_file("s3://bucket/raw/customers.parquet", db, "customers", schema="stg",
            if_exists="replace", primary_key="customer_id")

# Excel with messy headers
import_file("Customer List.xlsx", db, "customer_list", schema="stg", normalize_columns=True)

# DDL only
print(generate_ddl("data/customers.csv", "mssql", "customers", schema="dbo", primary_key="account_no"))

# insert into an existing table
insert_file("s3://bucket/incoming/orders.csv", db, "orders", schema="dbo")
```

### `if_exists`

| Value | Table doesn't exist | Table exists |
| --- | --- | --- |
| `"fail"` (default) | created | **error**, data is safe |
| `"append"` | created | rows are added |
| `"truncate"` | created | emptied, structure kept, then filled |
| `"replace"` | created | DROPped, recreated from the data, then filled |

### How the DDL is derived

- **Column types** come from the data. Parquet keeps its native types. CSV types are detected from the **whole file**, not a sample, so a wrong guess can't fail halfway through an insert. Excel uses each cell's native type.
- **Codes/IDs stay text.** CSV columns that look numeric but start with `0` (zip code `"02108"`) or are ≥ 16 digits long (account/card numbers) are stored as text. Tune the digit threshold with `text_id_digits` (`None` disables it).
- **Text lengths** are measured from the data: `max_len × 1.2`, rounded up to 50/100/255/500/1000/2000/4000. Anything longer becomes `NVARCHAR(MAX)` / `TEXT`. MySQL: more than 255 characters becomes `TEXT`, so wide tables stay under the 65 KB row limit.
- **List/struct columns** (from Parquet/JSON) are stored as JSON: `JSONB` on Postgres, `JSON` on MySQL, `NVARCHAR(MAX)` on SQL Server.
- **Overrides**: `primary_key=`, `sql_types={"account_no": "CHAR(16)"}`, `unicode=False` (mssql uses `VARCHAR`), `not_null_from_data=True`, `varchar_padding=`, `varchar_min=`.

### Source reading options

These options apply to `import_file`, `insert_file`, `generate_ddl`, and `FileSource`.

| Option | Example | Purpose |
| --- | --- | --- |
| `normalize_columns` | `True` | `"Account No."` → `account_no`, `"Customer Name"` → `customer_name` |
| `rename` | `{"Account No": "account_id"}` | rename specific columns |
| `columns` | `["account_no", "city"]` | keep only some columns |
| `where` | `"country = 'US'"` | filter rows (DuckDB SQL) |
| `sheet` | `"Sheet2"` or `1` | Excel sheet |
| `csv_options` | `{"delim": ";", "encoding": "latin-1"}` | DuckDB `read_csv` options |
| `source_types` | `{"region_code": "VARCHAR"}` | force types while reading |
| `s3` | `S3Client(endpoint_url=...)` | S3 client for `s3://` sources |

### Inserts: fast & safe to re-run

- Data is streamed in batches (default 100k rows on Postgres, 50k on SQL Server, 20k on MySQL; change with `batch_size=`). RAM stays small even for multi-GB files.
- The fastest path per database: **COPY** (Postgres), **fast_executemany** (SQL Server), **multi-row INSERT** (MySQL). In local tests: ~200–300k rows/s on Postgres and ~65k rows/s on MySQL.
- **Resumable:** every batch is committed and progress is saved to `.dbhop/import_<schema>.<table>.checkpoint.json`. If the process dies (VPN, network), run the same command again and it **resumes from the last batch with no duplicate rows**. The checkpoint is removed on success. Use `restart=True` / `--restart` to start from zero.
- `on_error="skip"`: a failing batch is retried row by row. Failing rows are logged to `….errors.jsonl`; the rest still loads.
- `verify=True` (default): afterwards, the table's row growth is compared with the number of inserted rows.
- **Inserting into an existing table:** columns are matched by name (case-insensitive). File columns missing from the table are ignored; table columns missing from the file get their DEFAULT/NULL. Both lists are printed so nothing is silent.

---

## Upsert & delete into SQL Server

For keeping a SQL Server table in sync with a Parquet snapshot:

```python
db = DatabaseClient.from_env("WAREHOUSE_DB", db_type="mssql")

# preview which columns will be synced, without touching the database
db.upsert_parquet("customers.parquet", "customers", key_column="customer_id", dry_run=True)

# MERGE: update matching rows, insert new ones
db.upsert_parquet("customers.parquet", "customers", key_column="customer_id")

# delete rows whose key appears in the parquet file
db.delete_by_keys("customers_removed.parquet", "customers", key_column="customer_id")
```

- Only columns present on **both** sides are synced; target columns missing from the Parquet file are left untouched. `created_at` / `updated_at` are never overwritten, and `updated_at` is set to `SYSDATETIME()` on update when the column exists.
- Data goes through a session temp table in chunks with `ROWLOCK`, so locks are short-lived and never escalate to the whole table.
- Resumable via a checkpoint file. Composite keys are supported (`key_column=["customer_id", "version"]`).
- Bad rows in a chunk are retried one by one and logged to `<checkpoint>.errors.jsonl` instead of failing the whole run.
- `upsert_parquet_parallel(db, ...)` / `delete_by_keys_parallel(db, ...)` split the work across processes. Start with 2–4 workers: parallel MERGEs on the same table can still block each other through shared index pages. Call them under `if __name__ == "__main__":` because they use multiprocessing.

---

## Progress & monitoring

Every operation reports progress per stage automatically: export, S3 upload/download, import, insert, upsert/delete (including parallel runs), and large-file scans. No setup is needed. There are three ways to see it:

### 1. Progress bars (terminal & Jupyter)

When run interactively, output uses [tqdm](https://github.com/tqdm/tqdm): percentage, counts, rate, and time remaining (ETA).

```text
▶ import: s3://bucket/raw/customers.csv -> "stg"."customers"
  download: 100%|██████████████████████████| 176M/176M [00:02<00:00, 80.2MB/s]
  detect CSV format: 100%|█████████████████████████████████████| [00:06<00:00]
  check code/ID columns: 100%|█████████████████████████████████| [00:00<00:00]
  ⋯ prepare table ...
  ✓ prepare table in 0:00:01
  insert:  35%|█████████▊                  | 700k/2.00M [00:04<00:09, 134k rows/s]
  ⋯ verify ...
  ✓ verify in 0:00:00
  2,000,000 rows loaded into "stg"."customers"
  ✔ finished import: s3://bucket/raw/customers.csv -> "stg"."customers" in 0:00:28
```

- Stages executed inside DuckDB (CSV detection, column checks, profiling, building the upsert index) use DuckDB's own progress percentage.
- In Jupyter, bars render as plain text. Install the `notebook` extra (`ipywidgets`) for widget bars.

### 2. Log lines (cron, nohup, `> file.log`)

When output isn't a terminal, the bars are replaced by periodic log lines (every 10 seconds by default), keeping log files clean:

```text
[09:57:37] insert 2,000,000 rows -> "stg"."big" [postgresql] | batch 100,000
[09:57:41] insert 500,000 rows/2,000,000 rows (25.0%) | 144,511 rows/s | ETA 0:00:10
[09:57:44] insert 1,000,000 rows/2,000,000 rows (50.0%) | 146,928 rows/s | ETA 0:00:07
[09:57:53] ✓ insert: 2,000,000 rows in 0:00:16 (128,932 rows/s)
```

### 3. Watching from another terminal: `dbhop status`

Every job writes its status and log to `~/.dbhop/`, so background processes (e.g. `nohup` on a server) can be monitored from any other terminal:

```bash
dbhop status                 # last 20 jobs
dbhop status --watch         # refresh every 2 seconds (Ctrl+C to exit)
dbhop status --running       # running jobs only
dbhop status 20261020-0957   # one job's details (an ID prefix is enough) + last 15 log lines
dbhop status --clean 30      # delete finished job history older than 30 days
```

```text
ID                    KIND             STATUS      PROGRESS                ETA      ELAPSED  NAME
20261020-100124…a22e  import           ⏳ running  insert 45.0% (900,000)  0:00:12  0:00:31  s3://raw/orders.csv -> "stg"."orders"
20261020-100048…3dac  upsert-parallel  ✔ done                                       0:10:02  dbo.customers <- customers.parquet
20261020-100050…0ce7    └ upsert       ✔ done                                       0:09:40  dbo.customers <- customers.parquet [rows 0-1,000,000]
20261020-100050…245b    └ upsert       ✔ done                                       0:10:01  dbo.customers <- customers.parquet [rows 1,000,000-2,000,000]
20261020-100102…b3a4  import           ✘ failed                                     0:00:01  data/orders.csv -> "stg"."orders"
```

| Status | Meaning |
| --- | --- |
| `⏳ running` | in progress |
| `✔ done` | finished |
| `✘ failed` | stopped by an error; the message & traceback are in the details/log |
| `⏹ interrupted` | stopped with Ctrl+C |
| `☠ dead` | the process vanished without reporting (`kill -9`, out of memory, server restart). For import/upsert, run the same command again to resume from the checkpoint |

Parallel upserts/deletes have a parent job that sums the progress of all workers (one total + ETA). Each worker is still visible as a child job (`└`).

From Python:

```python
from dbhop import list_jobs, get_job

for j in list_jobs(running_only=True):
    st = j["stage"]
    print(j["name"], st["name"], st["pct"], st["eta_sec"])

get_job("20261020-0957")["stages"]   # every stage: done, total, rate, eta_sec, elapsed_sec
```

### Settings

| Env var | Default | Purpose |
| --- | --- | --- |
| `DBHOP_PROGRESS` | automatic | `bar` (force tqdm), `line` (force log lines), `off` (no output; status files are still written) |
| `DBHOP_PROGRESS_INTERVAL` | `10` | seconds between progress lines in `line` mode |
| `DBHOP_HOME` | `~/.dbhop` | folder for job history (`jobs/`) & logs (`logs/`) |

`verbose=False` on any function silences screen output; the job is still recorded and visible in `dbhop status`.

---

## CLI

Run `dbhop <command> --help` for every option.

```bash
# check this machine's setup
dbhop doctor --db-env WAREHOUSE_DB --check-s3

# export a table
dbhop export-table --db-env WAREHOUSE_DB \
    --table orders --schema dbo \
    --dest "s3://my-bucket/backup/{table}/dt={date}/"

# export a table with a filter & selected columns
dbhop export-table --db-env WAREHOUSE_DB \
    --table orders --schema dbo \
    --columns order_id,region,total_amount \
    --where "region = :region" --param region=EU \
    --bucket my-bucket --prefix "by_region/{date}" --filename "orders_eu.parquet"

# export a query (inline or from a .sql file); comma-separated values -> IN clause
dbhop export-query --db-env WAREHOUSE_DB \
    --query "SELECT * FROM dbo.orders WHERE region IN :regions" \
    --param regions=EU,APAC --name sample --dest "s3://my-bucket/report/"
dbhop export-query --db-env WAREHOUSE_DB --query-file sql/summary.sql \
    --name summary --dest "s3://my-bucket/report/{date}/"

# upload local files / folders
dbhop upload result.parquet --dest "s3://my-bucket/manual/"
dbhop upload exports/ --bucket my-bucket --prefix "manual/batch_01"

# browse S3 & the database
dbhop ls                                       # list buckets
dbhop ls s3://my-bucket/export/                # list a folder
dbhop tables --db-env WAREHOUSE_DB --schemas     # list schemas
dbhop tables --db-env WAREHOUSE_DB --schema dbo  # list tables + estimated rows & size
dbhop tables --db-env WAREHOUSE_DB --schema dbo --columns-of orders
```

More export options: `--order-by`, `--limit`, `--local-dir`, `--keep-local`, `--skip-if-exists`, `--chunksize`, `--compression {snappy,zstd,gzip,none}`, `--s3-endpoint`, `--db-url`, `--db-type`.

**Import (file / S3 → DB):**

```bash
# look inside a file: columns, types, NULL counts, max length, sample rows
dbhop inspect "s3://my-bucket/raw/Customer List.xlsx" --normalize-columns

# DDL only: without a DB connection (--db-type), or take the type from env (--db-env)
dbhop ddl data/customers.csv --db-type mssql --schema dbo --pk account_no \
    --normalize-columns --sql-type "account_no=CHAR(16)" -o create_customers.sql

# one-step import
dbhop import s3://my-bucket/raw/customers.parquet --db-env WAREHOUSE_DB \
    --schema stg --table customers --if-exists replace --pk customer_id

# every file in a folder combined into one table
dbhop import "s3://my-bucket/raw/orders/" --db-env WAREHOUSE_DB --schema stg --table orders_all

# each file in a folder becomes its own table
dbhop import s3://my-bucket/reference/ --each --db-env WAREHOUSE_DB --schema ref \
    --normalize-columns --if-exists replace

# insert into an existing table
dbhop insert data/orders.csv --db-env WAREHOUSE_DB --schema dbo --table orders \
    --truncate --rename "Order No=order_id" --on-error skip

# semicolon-delimited CSV + Windows encoding
dbhop import data/legacy.csv --db-env DB --csv-opt "delim=;" --csv-opt encoding=latin-1
```

More import options: `--dry-run`, `--restart`, `--batch-size`, `--no-verify`, `--sheet`, `--columns`, `--where`, `--type col=VARCHAR`, `--no-scan`, `--no-unicode`, `--padding`, `--min-len`, `--not-null-from-data`.

Scheduling with cron on a server (`cd` first so `.env` is found):

```cron
0 2 * * * cd /opt/etl && .venv/bin/dbhop export-table --db-env WAREHOUSE_DB --table orders --schema dbo --dest "s3://my-bucket/daily/{table}/dt={date}/" >> /var/log/dbhop.log 2>&1
```

A long job that must survive an SSH disconnect, monitored from another session:

```bash
nohup dbhop import s3://my-bucket/raw/orders.parquet --db-env WAREHOUSE_DB --schema stg > import_orders.log 2>&1 &
dbhop status --watch
```

---

## API reference

### `DatabaseClient`

```python
DatabaseClient("postgresql://user:pass@host:5432/db")          # URL
DatabaseClient(db_type="mssql", conn_params={                  # dict (safe for passwords with special characters)
    "host": "10.0.0.5", "port": 1433, "user": "u", "password": "p@ss", "database": "warehouse",
    "driver": "ODBC Driver 17 for SQL Server",                 # optional, auto-detected by default
})
DatabaseClient.from_env("WAREHOUSE_DB", db_type="mssql")       # from env vars
```

`db_type` aliases are accepted: `postgres` → `postgresql`, `sqlserver` → `mssql`, `mariadb` → `mysql`.

| Method | Purpose |
| --- | --- |
| `get_query(sql)` / `get_query_param(sql, params)` | SELECT → DataFrame |
| `execute_query(sql, bind)` / `executemany_query(sql, rows)` | Run DML/DDL directly |
| `export_table_to_parquet(table, file_path=None, schema=None, columns=None, where=None, params=None, order_by=None, limit=None, compression="snappy", chunksize=100_000, count_total=True)` | Table → Parquet |
| `export_query_to_parquet(sql, file_path, params=None, ...)` | Query → Parquet (streaming) |
| `build_table_query(table, schema, columns, where, order_by, limit)` | SELECT with dialect-correct quoting |
| `insert_dataframe(df, table, if_exists="append")` | DataFrame → table (any DB) |
| `insert_dataframe_fast(df, table, schema="dbo", if_exists="fail")` | Faster variant for SQL Server |
| `upsert_parquet(parquet, table, key_column, ...)` | MERGE Parquet → SQL Server, resumable |
| `delete_by_keys(parquet, table, key_column, ...)` | DELETE by the keys in a Parquet file, resumable |
| `list_schemas()` | Schemas (excluding system schemas) |
| `list_tables(schema=None, like=None)` | DataFrame of tables + **estimated** rows & size (from DB statistics, fast) |
| `table_columns(table, schema=None)` | DataFrame of columns: name, type, nullable |
| `count_rows(table, schema=None, where=None, params=None)` | Exact row count (`COUNT(*)`) |
| `table_exists(table, schema=None)` / `create_schema_if_missing(schema)` | Existence checks / schema creation |
| `inspector()` | SQLAlchemy inspector |

Module functions: `upsert_parquet_parallel(db, ...)`, `delete_by_keys_parallel(db, ...)`, `detect_mssql_driver()`, `detect_postgres_driver()`.

**About `where` / `order_by`**: they are raw SQL fragments. Write them in your own code, **never** from user input. Always pass dynamic values through `:param` + `params={...}`. A list/tuple/set value expands into an `IN` clause automatically.

### `S3Client`

```python
S3Client(bucket=None, endpoint_url=None, access_key=None, secret_key=None,
         region=None, addressing_style=None, signature_version="s3v4")
```

| Method | Purpose |
| --- | --- |
| `upload_file(local, key=None, bucket=None, prefix=None, dest=None, metadata=None, verify=True, ...)` | Upload one file (automatic multipart), returns `s3://...` |
| `upload_dir(local_dir, bucket=None, prefix=None, dest=None, pattern="*")` | Upload a folder recursively |
| `download_file(key, local_path, bucket=None)` | Download |
| `exists(key, bucket=None)` | Does the object exist? |
| `list(prefix="", bucket=None)` | List objects → `[{key, size, last_modified}]` |
| `delete(key, bucket=None)` | Delete an object |
| `presigned_url(key, bucket=None, expires=3600)` | Temporary download link |
| `list_buckets()` / `bucket_exists(bucket)` / `create_bucket(bucket)` | Check & create buckets (`PermissionError` without permission) |
| `resolve(dest, bucket, prefix, filename)` | Compute `(bucket, key)` using the destination rules above |

Helpers: `parse_s3_uri("s3://b/k") -> ("b", "k")`, `join_key("a/", "/b", "c") -> "a/b/c"`.

### Export DB → S3

```python
export_table_to_s3(db, s3, table_name, dest=None, bucket=None, prefix=None, filename=None,
                   schema=None, columns=None, where=None, params=None, order_by=None, limit=None,
                   local_dir=None, keep_local=False, skip_if_exists=False,
                   compression="snappy", chunksize=100_000, metadata=None, verbose=True) -> dict

export_query_to_s3(db, s3, query, dest=None, bucket=None, prefix=None, filename=None,
                   name="query", params=None, ...) -> dict

export_tables_to_s3(db, s3, tables, continue_on_error=True, **shared_kwargs) -> list[dict]
```

- Result `status`: `uploaded`, `skipped` (because of `skip_if_exists`), or `failed` (only in `export_tables_to_s3`).
- Objects get `source`, `rows`, and `exported-at` metadata, plus anything passed in `metadata=`.

### Import file / S3 → DB

```python
import_file(source, db, table_name=None, schema=None, if_exists="fail",
            primary_key=None, sql_types=None, create_schema=True,
            scan_strings=True, varchar_padding=1.2, varchar_min=50, unicode=True,
            not_null_from_data=False, batch_size=None, on_error="raise",
            checkpoint_path=None, restart=False, verify=True, dry_run=False,
            verbose=True, **source_options) -> dict
# -> {table, created, ddl, rows_source, inserted, failed, verified, rows_per_sec, error_log, ...}

insert_file(source, db, table_name, schema=None, truncate=False, batch_size=None,
            on_error="raise", checkpoint_path=None, restart=False, verify=True,
            **source_options) -> dict

import_files(sources, db, schema=None, table_names=None, continue_on_error=True,
             s3=None, **import_file_kwargs) -> list[dict]

generate_ddl(source, db_type, table_name=None, schema=None, primary_key=None,
             sql_types=None, if_not_exists=False, scan_strings=True,
             not_null_from_data=False, varchar_padding=1.2, varchar_min=50,
             unicode=True, **source_options) -> str
plan_columns(source, db_type, ...) -> list[ColumnSpec]          # editable column plan
render_create_table(columns, table_name, db_type, schema=None, primary_key=None,
                    if_not_exists=False) -> str
map_type(duck_type, db_type, max_len=None, ...) -> str           # DuckDB type -> target SQL type
```

`db_type` may be a string (`"mssql"`, `"postgresql"`, `"mysql"`) or a `DatabaseClient`.

`FileSource(source, s3=None, file_format=None, sheet=None, csv_options=None, source_types=None, columns=None, rename=None, normalize_columns=False, where=None, text_id_digits=16, download_dir=None)` provides:

- `.columns`, `.column_names`
- `.row_count()`, `.profile()`
- `.print_info()`, `.preview(n)`
- `.iter_batches(n)`

Use `with FileSource(...) as src:` so temporary files are cleaned up. One `FileSource` can be passed to both `generate_ddl` and `import_file`, so an S3 file is downloaded only once.

---

## Driver detection (laptop vs. server)

So the same code runs on any machine, drivers are chosen at runtime.

**SQL Server (pyodbc)**, in order of priority:

1. `conn_params["driver"]` or env `<PREFIX>_DRIVER`, for that connection only
2. env `MSSQL_ODBC_DRIVER`, for every mssql connection on the machine
3. Auto-detected from `pyodbc.drivers()`: **Driver 18 → 17 → 13.1 → 13 → FreeTDS → "SQL Server"**

Connection options adapt to the driver:

- Driver 17/18: `Encrypt=no;TrustServerCertificate=yes`. Driver 18 requires TLS by default, which often fails against internal servers. Set `MSSQL_ENCRYPT=yes` if your server really uses TLS.
- FreeTDS: `TDS_Version=7.4` (change via `MSSQL_TDS_VERSION`).
- The port is written as `SERVER=host,port`, the form Microsoft's driver actually honors.

**PostgreSQL**: uses `psycopg2` if installed, otherwise `psycopg` (v3). `postgresql://` URLs are pinned to the installed driver, which matters because SQLAlchemy 2.1+ defaults to psycopg 3 while 2.0 defaults to psycopg2.

**MySQL**: always `pymysql` (pure Python, nothing to compile). `mysql://` URLs are rewritten to `mysql+pymysql://`.

Example: a laptop with Driver 18 and a server with only Driver 17 both work with no change to code or `.env`. Check which driver is used with `dbhop doctor`.

---

## Troubleshooting

| Symptom | Cause / fix |
| --- | --- |
| `No SQL Server ODBC driver is installed` | Install a driver ([Installation](#odbc-driver-sql-server-only)) or set `MSSQL_ODBC_DRIVER` |
| `SSL Provider: certificate verify failed` | Make sure `MSSQL_ENCRYPT` is not set to `yes` |
| `Login timeout expired` | Check host/port, firewall/VPN, and `SERVER=host,port` (see `doctor`) |
| `No module named 'psycopg'` / `'psycopg2'` | `pip install "dbhop[postgres]"` |
| `No bucket specified` | Pass `dest=`/`bucket=`, or set `S3_BUCKET` in `.env` |
| `SignatureDoesNotMatch` / `InvalidAccessKeyId` | Wrong credentials, or a missing/wrong S3 endpoint |
| MinIO/Ceph upload fails with `NoSuchBucket` although the bucket exists | Try `S3_ADDRESSING_STYLE=path` (the default when `S3_ENDPOINT` is set) |
| `Schema of chunk N is incompatible` | A column has mixed types across chunks. Increase `chunksize` or `CAST` in the query |
| Integer columns become float in Parquet | pandas reads integer columns containing NULLs as float |
| Import: `Table ... already exists (if_exists='fail')` | Intentional, to keep data safe. Choose `append` / `truncate` / `replace` |
| Import: `Checkpoint ... comes from a different source/columns` | A previous run with another file was interrupted. Use `restart=True` / `--restart` |
| Import: `No source columns match the target table` | File column names differ from the table. Use `normalize_columns` or `rename` |
| Import CSV: garbled characters (`Ã©`) / read errors | Probably a Windows-encoded file. Use `csv_options={"encoding": "latin-1"}` |
| Import CSV: all columns merged into one | The delimiter isn't a comma. Use `csv_options={"delim": ";"}` |
| Import Excel: long ID numbers lose their last digit | Excel stores at most 15 significant digits. Store such columns as **Text** in Excel |
| No progress shown / messy output in log files | Force a mode: `DBHOP_PROGRESS=line` (logs) or `bar` (terminal) |
| `dbhop status` shows `☠ dead` | The process vanished (kill/OOM/restart). Run the same command again; import/upsert resume from the checkpoint |
| SQL Server inserts are very slow | There's an `NVARCHAR(MAX)` column; fast_executemany is much slower for MAX columns. Use `sql_types` with a fixed size |

---

## Development

```bash
git clone <repository-url> && cd dbhop
python -m venv .venv && source .venv/bin/activate
pip install -e ".[all,dev]"
pytest
```

The test suite runs without any database server or S3 endpoint.

Releasing to PyPI:

```bash
# bump __version__ in src/dbhop/__init__.py and update CHANGELOG.md, then:
rm -rf dist && python -m build
twine check dist/*
twine upload --repository testpypi dist/*   # optional dry run on TestPyPI
twine upload dist/*
```

---

## License

[MIT](LICENSE)
