conduto

The pipe that carries your data from source to destination.

🇺🇸 English · Português (BR)

conduto is a CLI that scaffolds a complete ELT project in a single command: credentials for both ends in the .env, YAML schemas for the tables, DDL in the target database and Dagster code (assets + schedules) ready to run.

What it's for: you have data in one database and need to keep it synchronized in another. Without conduto that means hand-writing the driver, credentials, cross-database type mapping, CREATE TABLE on the target, incremental reads by watermark, scheduling and orchestration. With conduto you answer a wizard and get a project that loads the data — incremental or full loads, on a schedule, plus an interface to trigger and monitor each table.

• Source and target: PostgreSQL, MySQL, SQL Server, ClickHouse, DuckDB and Delta Lake
• Orchestration: Dagster (independent assets, per-table schedules, web UI)
• Install: pip install "conduto[postgresql]" (drivers per database)
• Code: github.com/joaopedrozg/conduto · Issues: open an issue

Contents

1. Installation
2. Quick start
3. The commands
4. The interactive wizard
5. What gets generated
6. How data is loaded
7. Configuration reference
8. Supported databases
9. Development

Installation

The base package installs only the CLI (Typer, Rich, Textual, Jinja2 and PyYAML). Database drivers come as extras per database — install only what you use:

pip install "conduto[postgresql]"   # psycopg
pip install "conduto[mysql]"        # pymysql
pip install "conduto[sqlserver]"    # pyodbc
pip install "conduto[clickhouse]"   # clickhouse-connect
pip install "conduto[duckdb]"       # duckdb
pip install "conduto[deltalake]"    # deltalake, pyarrow and boto3

pip install "conduto[all]"          # every database

Same thing with uv:

uv tool install "conduto[all]"

# or run it without installing it for good
uvx --from "conduto[postgresql]" conduto init my_project

Requirements: Python 3.10+ (the package declares support for 3.10–3.14) and uv on your PATH — conduto uses it to set up the generated project's environment.

Without the extra, the CLI still works fine (conduto --help, ddl, schedules, docs...). When you pick a database whose driver is missing, conduto asks whether to install it right away (via uv or pip) and prints the manual command if it fails. Projects generated by conduto init get the database driver in their own uv add — the extra is only for the CLI's environment.

Linux: PEP 668 and the "environment required" error

Debian/Ubuntu, Fedora, Arch and Homebrew on macOS mark the system Python as externally-managed (PEP 668): pip install outside a venv is refused. This does not affect uvx, uv tool or pipx users — they already run in their own venv.

When it happens, conduto detects the marker before installing and shows both ways out:

uvx --from "conduto[postgresql]" conduto   # isolated venv, system untouched

pip install --break-system-packages psycopg[binary]   # only if you allow it

The prompt "Install on the system Python anyway?" defaults to no — nothing is broken without explicit consent.

Quick start

# 1. Creates the project: connections in .env, schemas/ and main.yml
conduto init my_project

# 2. Generates the table DDL and applies it to the target database
conduto ddl --apply

# 3. Generates the schedules and the Dagster code
conduto schedules

# 4. Starts the Dagster server (http://localhost:3000)
conduto dagster

# 5. Web documentation of the project (http://localhost:8000)
conduto docs

Check the version and the help:

conduto --version
conduto --help
conduto [COMMAND] --help

What conduto init asks

The flow is interactive and guided — you never type database or table names by hand:

1. Source database — default port and user change per database
2. Source credentials — host, port, user and password
3. Connection test — if it fails, retype them or continue anyway
4. Databases on the source server — conduto lists them and you pick one
5. Target database and credentials — same flow, plus picking the target schema (with an option to create a new database/schema)
6. Schema mode — generate automatically (you check the source schemas and tables) or configure manually (generates three examples for you to edit)
7. Schedules — whether to generate each table's schedule automatically (default: hourly) and the Dagster code
8. Dagster server — whether to start it now and generate the run_dagster.ps1 / run_dagster.sh scripts

Inside an existing uv project? If the current directory already has a pyproject.toml (for example after uv add conduto), conduto adapts: it writes .env, main.yml and schemas/ straight into the current project and adds only the missing dependencies — no subfolder, no uv init. In that case use uv run conduto init (the project name becomes optional).

Generating schemas automatically

After both connections are tested, in generate automatically mode conduto:

• lists the source database schemas and opens a selection screen with a search filter — check any number with space, all visible ones with a (Select all) and clear with l;
• then shows the tables of the checked schemas using the same search/check screen — tables that already have schemas/<table>.yml show up in amber as "already exists" (regenerating overwrites);
• reads the columns from the database (types, PK, FK, unique, default, nullable) and writes schemas/*.yml and main.yml.

Your checks are what defines the source schema of each table — no other schema prompt shows up afterwards. In manual mode, conduto asks the source schema once and generates the clientes, pedidos and produtos examples.

Adding a new table

Create the schema with just the name (or add the path to main.yml) and let conduto fill in the columns:

# schemas/my_table.yml
table: my_table

conduto inferir                      # infers every schema without columns
conduto inferir --tabela my_table
conduto schedules                    # generates schedule and Dagster code for the new one

The command reads the source credentials from the .env, queries the database and writes the columns: while preserving whatever already exists (description, schedule etc.).

The commands

Global flags: --lang pt|en (language for the run) and --version.

The command names stay in Portuguese (inferir, ddl, schedules...) — only the messages, prompts and help texts are translated.

Usage examples

# only write the DDL to a file, never touching the database
conduto ddl --no-apply --output ddl.sql

# apply straight away, without prompting (handy for scripts/CI)
conduto ddl --apply

# work in another directory
conduto schedules --dir path/to/project
conduto docs --port 9000 --no-open

Web documentation

conduto docs                  # opens http://localhost:8000
conduto docs --port 9000      # specific port
conduto docs --no-open        # don't open the browser

The page shows the project overview, the file tree, the .env connections (passwords masked), the per-database particularities, the tables with their schemas, the schedules, the generated DDL, the environment and the list of commands.

The interactive wizard

In the terminal, conduto init and conduto ddl run inside a TUI screen (Textual):

• Side menu of steps with the state of each one: current (● blue), done (✓ green), skipped (— gray) or pending (○);
• Review: clicking a completed step shows what was answered there, without rerunning the flow;
• Log panel with what the command is doing, without flashing old logs on every step change;
• Logs in SQLite: all output is kept in ~/.conduto/registros.db and opens with F3, with timestamp, step and scrolling — the output is also replayed in the terminal when the shell closes;
• Footer shortcuts: F2 toggles the step menu, F3 opens the logs, esc goes back;
• Colors are status: green = success, amber = attention, red = error, blue = information, gray = neutral;
• Loading widgets (spinner and progress bar) on slow operations such as the connection test and applying the DDL.

Without an interactive terminal (CI, pipe) or with the CONDUTO_SEM_TUI variable set, everything falls back to the classic numbered prompts in the terminal — same behaviour, no screen.

Logs live at ~/.conduto/registros.db; point CONDUTO_REGISTROS somewhere else to change that.

What gets generated

my_project/
├── .env                        # source and target credentials
├── main.yml                    # manifest: table list + overall schedule
├── schemas/                    # one YAML per table
│   ├── clientes.yml
│   ├── pedidos.yml
│   └── produtos.yml
├── definitions.py              # entry point for `dagster dev`
├── conduto_dagster/            # generated Dagster code
│   ├── __init__.py
│   ├── etl.py                  # connection, read and load of the tables
│   └── definitions.py          # assets and schedules built from the YAMLs
├── run_dagster.ps1             # Windows
├── run_dagster.sh              # Linux/macOS
└── pyproject.toml              # uv environment (pyyaml, jinja2, dagster, dagster-webserver)

Outside a uv project, this structure is created inside my_project/. Inside an existing uv project, the files go to the current directory. The project is created without an src/ folder (uv init --bare) — scripts and Dagster code live at the root.

.env — credentials

DB_ORIGEM_TYPE=postgresql
DB_ORIGEM_HOST=localhost
DB_ORIGEM_PORT=5432
DB_ORIGEM_NAME=postgres
DB_ORIGEM_SCHEMA=public
DB_ORIGEM_USER=postgres
DB_ORIGEM_PASSWORD=postgres

DB_DESTINO_TYPE=postgresql
DB_DESTINO_HOST=localhost
DB_DESTINO_PORT=5432
DB_DESTINO_NAME=postgres
DB_DESTINO_SCHEMA=public
DB_DESTINO_USER=postgres
DB_DESTINO_PASSWORD=postgres

# Optional: rows per load batch (default 20000)
# CONDUTO_LOTE=50000

# Optional (PostgreSQL): load statement_timeout in ms (0 = no limit)
# DB_DESTINO_STATEMENT_TIMEOUT=0

Variable names stay in Portuguese: DB_ORIGEM_* holds the source and DB_DESTINO_* the target.

DB_ORIGEM_SCHEMA is the default source schema — the ETL only uses it when the table has no source_schema of its own in the YAML.

Important: the .env holds credentials and must not be committed.

main.yml — manifest

version: "1.0"
project: my_project

# Schedule of the whole model (every table, in manifest order)
schedule:
  cron: "0 * * * *"

tables:
  - path: "schemas/clientes.yml"
  - path: "schemas/pedidos.yml"
  - path: "schemas/produtos.yml"

schemas/*.yml — tables

table: clientes
schema: public              # schema in the TARGET (what conduto ddl uses)
source_schema: public       # schema in the SOURCE (where the ETL reads from)
description: "Customer registry table"
schedule:
  cron: "0 * * * *"
  mode: incremental         # incremental or full
  incremental_column: criado_em
  full_load: false          # true forces a full load on the next run
  truncate: false           # true wipes the target table before loading
columns:
  - name: id
    type: integer
    primary_key: true
    nullable: false
  - name: email
    type: varchar(255)
    unique: true
  - name: criado_em
    type: timestamp
    default: CURRENT_TIMESTAMP

Two schema keys with different jobs:

The source one is stored per table because a table can live in one schema and another in a different one (Person, HumanResources, dbo...). Without it the ETL would read everything from the single DB_ORIGEM_SCHEMA in the .env and fail with Invalid object name. In older projects without the key, DB_ORIGEM_SCHEMA still applies as a fallback.

The three examples cover common modelling patterns:

No dependencies between tables

foreign_key is model documentation only — nothing in the project turns it into a constraint:

• conduto ddl never emits a FOREIGN KEY in the CREATE TABLE, on any target;
• the Dagster code never creates deps between assets;
• main.yml is never reordered by FK topology.

So if pedidos references clientes, you can still load pedidos on its own, without clientes existing or having run first. Referential integrity stays with whoever writes to the source.

Custom types

Types that don't exist in every database — ltree, citext, hstore, tsvector, inet, geometry, interval, hierarchyid, year, PostgreSQL arrays etc. — follow a rule per target: the type is kept where it exists natively (PostgreSQL, plus geometry/hierarchyid in SQL Server) and degrades to text the target accepts in the other cases (text/nvarchar(max)/String/varchar/string). That way the CREATE TABLE never fails because of a type.

When the rule isn't the one you want, declare the equivalent on the column with types: — it beats the rule table:

columns:
  - name: localizacao
    type: geometry            # default rule per target
    types:                    # override: only for the listed targets
      mysql: point
      deltalake: binary

If a type has no rule at all it passes through as-is and conduto warns you — a sign that a types: declaration is worth it.

How data is loaded

The generated Dagster code (conduto_dagster/) reads main.yml and the schemas/*.yml at runtime: changing a schema's schedule key changes the asset/schedule on the next reload, with nothing to regenerate.

Each table is an independent asset, with the mode defined in its own YAML:

How loading happens per target:

• PostgreSQL: COPY — direct streaming COPY (SELECT ...) TO STDOUT → COPY ... FROM STDIN when source and target are both PostgreSQL (Python only forwards bytes), and COPY FROM STDIN in pre-serialized blocks from other sources. On an already populated table with a PK, the target gets an upsert on the primary key so incremental loads don't raise UniqueViolation.
• ClickHouse: native batched INSERT.
• DuckDB: rows via Arrow + INSERT SELECT (DuckDB's executemany is slow, one row at a time).
• Delta Lake: write_deltalake.
• SQL Server / MySQL and the rest: batched INSERT, with fast_executemany (bulk operation) on SQL Server and an automatic fallback if the driver refuses.

The source is always read in batches (fetchmany) with a producer/consumer thread pair — memory stays stable even on huge tables.

Running it

# in the browser (http://localhost:3000): select the assets and click "Materialize"
uv run dagster dev

# or through the generated scripts
.\run_dagster.ps1      # Windows
./run_dagster.sh       # Linux/macOS

# or straight from conduto
conduto dagster
conduto dagster --dir path/to/project

dagster dev requires the dagster-webserver package; conduto installs it along with the other dependencies and, if it's missing in an existing project, installs it before starting the server (uv add dagster-webserver). While the server boots, conduto shows an animated status and tells you when it's up. If the conduto_dagster/ code doesn't exist yet it's generated on the spot and the [tool.dagster] block is added to pyproject.toml (recent Dagster versions require it).

Load performance

• CONDUTO_LOTE in the .env: rows per copy batch (default 20000). On heavy loads 50000 usually pays off.
• DB_DESTINO_STATEMENT_TIMEOUT (PostgreSQL, in ms; 0 = no limit): avoids hitting the server's statement_timeout during COPY. Use the session pooler (port 5432) or a direct connection — the transaction pooler (6543) doesn't allow session SET.
• For big loads, drop non-essential indexes/constraints from the target table before a full load and recreate them afterwards — COPY gets much faster without them.

Configuration reference

Environment variables

Language

conduto detects the machine's language (Portuguese by default, with English support) and uses that preference for messages, prompts, help and labels. Detection order:

1. CONDUTO_LANG
2. Locale variables (LANG, LC_ALL, LC_MESSAGES) and Python's locale
3. Windows display language

To force it for a single run: conduto --lang en init my_project.

--lang applies to the command's flow; --help follows automatic detection and can be forced with CONDUTO_LANG=en conduto --help.

The schedule key (per YAML schema)

conduto schedules infers the incremental update column (preferring updated_at/atualizado_em, then created_at/criado_em and finally any temporal column), writes the key into the schemas and keeps the overall model schedule in main.yml. Values you edited by hand are preserved on regeneration — only missing keys get the default.

Supported databases

Particularities applied automatically in the CLI flow and in the DDL:

• ClickHouse: MergeTree ENGINE/ORDER BY and no constraints;
• Delta Lake: no constraints in the CREATE TABLE;
• MySQL: database == schema;
• SQL Server: automatic download and install of the ODBC driver.

SQL Server ODBC driver

pyodbc needs the native driver on the system. If the connection fails because of it, conduto init offers Install driver automatically — the credentials you typed stay in memory only and the connection test reruns by itself afterwards. You can also install it directly:

conduto install-sqlserver-driver

Works on Windows (winget or MSI), Linux (apt) and macOS (Homebrew). On Windows, if the terminal isn't elevated, conduto brings up the UAC confirmation in front; with a pending reboot, the install is blocked until you restart. There's also a standalone script for manual/offline installs:

.\scripts\install-sqlserver-odbc.ps1 -DownloadOnly -OutFile .\msodbcsql18.msi   # download only
.\scripts\install-sqlserver-odbc.ps1                                            # download and install

Supported versions: 18 (default) and 17 (-Version 17). Official documentation: Download ODBC Driver for SQL Server.

Development

This README also exists in Portuguese, in README.md — an absolute link on purpose, because this is the file PyPI renders (readme = "README.en.md" in pyproject.toml), and relative links would break there. When you edit one, update the other; CI fails if the two drift apart.

uv sync --all-extras   # project + drivers for every database + dev (pytest)
uv run pytest -q       # tests
uv run python scripts/check_readme_sync.py   # checks both READMEs
uv build               # package it

CI (.github/workflows/ci.yml) runs on every push/PR: installs with uv sync --all-extras, smoke-tests the CLI, runs the tests and builds the package.

Publishing a new version

Every merge/push to main triggers the Publish to PyPI workflow, which computes the version from the last one published on PyPI (patch bump; if the PR already bumped pyproject.toml beyond the published one, that version is used), publishes and creates the vX.Y.Z tag. The bump is applied only in the workflow's working tree — it never lands on main.

For a manual release (patch, minor or major): Actions → Publish to PyPI → Run workflow and pick the bump type.

License

MIT