I migrated a Fivetran pipeline to dltHub for $0.65

TL/DR: I replaced one daily Fivetran MySQL sync with dltHub in about an hour. OpenCode running Minimax M3 (because multimodal and still cheap 😅) consumed $0.65 in tokens. I gave it five manual inputs. One was actual infrastructure work. The other four were decisions and verification.

The old connection pulled 12 CMS backend (our Drupal uses MySQL) tables from a webserver through an SSH tunnel into Snowflake. It worked. And it was billed via monthly-active-rows 🙄 And it felt like a reasonable candidate for a small experiment: could I recreate the operational behavior with dltHub, use OpenCode as the implementation environment, and stay in a range where an hour of AI usage costs less than a coffee in Zurich?

Turns out... yes. With caveats, obviously.

The baseline

I first had to log in to Fivetran for the first time in months (it is a reliable platform after all) to get the connector ID. Then I used the Fivetran MCP to inspect its MySQL configuration and how it connects to the webserver through an SSH tunnel. Its selected 12 tables are all part of a Drupal CMS, and it runs daily.

I did not want a generic MySQL ingestion example. I wanted the existing job's behavior: same source scope, data landing in Snowflake, incremental loads where I could justify them, and a downstream task triggered (❄️ Cortex Search refresh) after a successful load.

Five manual inputs

I did not sit back and accept generated code blindly. The migration needed five interventions.

1. Add a new SSH key: This was the major manual task. I created a dedicated key pair and added its public key to the servive user's authorized_keys file on the webserver. I also captured the server's host key while I was there so the new job can pin it instead of trusting whatever answers on port 22.
2. Decide source scope: The first profile discovered 346 tables and roughly 6.2 million rows. That was technically fun and operationally wrong. The previous Fivetran configuration had 12 enabled tables, because those are what I actually use. So I told OpenCode to match that scope exactly.
3. Verify the plan: I checked the generated design against the source schema and test suite. In particular, I wanted primary keys reflected from MySQL, a conservative cursor whitelist (for incremental loads: I didn't define the PK and cursor for each of the 12 tables individually but rather used a dltHub skill to annotate sources and identify usually useful cursors), and cleanup of the SSH tunnel.
4. Add a downstream task: Fivetran stopped after the load. I asked OpenCode to trigger the existing Snowflake Cortex Search refresh task after dlt completes. The new pipeline now does a little more work than its predecessor 😎
5. Verify the run: After everything was implemented and tested locally, I had to point OpenCode to a modification of the pipeline necessary for it to run on dltHub (cf. below).

What dltHub generated

dlt's sql_database source doesn't open SSH tunnels itself. Hence, my new job opens a pinned tunnel, then creates a SQLAlchemy engine through its ephemeral local port, and passes that engine to dlt.

def get_ssh_tunnel(ssh_credentials):
    private_key = paramiko.Ed25519Key.from_private_key(
        io.StringIO(ssh_credentials["private_key"])
    )
    return ParamikoTunnel(
        hostname=ssh_credentials["server_ip_address"],
        username=ssh_credentials["username"],
        private_key=private_key,
        host_key=parse_ssh_host_key(ssh_credentials["host_key"]),
        target_address=("127.0.0.1", 3306),
    )

The database has composite primary keys everywhere. I reflected MySQL metadata and only configured incremental merge for tables with a primary key plus one of three mutation-oriented columns:

SAFE_CURSOR_PRIORITY = (
    "changed",
    "content_translation_changed",
    "revision_timestamp",
)

def apply_resource_hints(table_name, resource, table):
    pk_columns = [column.name for column in table.primary_key.columns]
    cursor = choose_cursor(table_name, [column.name for column in table.columns])
    if cursor is None or not pk_columns:
        resource.apply_hints(write_disposition="replace")
        return

    resource.apply_hints(
        write_disposition="merge",
        primary_key=pk_columns,
        incremental=dlt.sources.incremental(
            cursor, on_cursor_value_missing="include"
        ),
    )

I deliberately rejected created, generic timestamp, and revision IDs as universal cursors. Drupal puts lots of time-related integers into its schema. A plausible name is not enough evidence for incremental correctness.

Two failures worth mentioning

The first remote run failed (and there goes my 100% pipeline-run-success-rate for the week 😤) because dltHub connected to Snowflake from an IP outside the account allowlist. I wrapped the job with my existing Snowflake proxy context used by other dltHub jobs. On the source-side of the pipeline, SSH and MySQL keep using raw TCP, so they are unaffected by those HTTP proxy variables.

dltHub Pro is here. So is my static-egress proxy VM.

How I kept IP‑locked Snowflake and Azure databases happy with a managed runtime that runs “wherever Python runs”

sfrtMartin Seifert

The first all-table run also exhausted SQLAlchemy's connection pool. I had given dlt hundreds of resources and long-lived MySQL reads. A larger pool only delayed the failure. NullPool fixed it by opening and closing a connection per source read:

return sa.create_engine(url, poolclass=sa.pool.NullPool)

One dependency issue also appeared here: The aforementioned sshtunnel calls Paramiko's removed DSSKey API, so I build a custom ParamikoTunnel instead. The smoke test found that before deployment, which is exactly why I keep smoke tests around.

Proof, then schedule

The scoped baseline run completed locally in 31 seconds. The second run restored state from Snowflake and completed in 16 seconds, confirming that dlt retained its cursors. The deployed dltHub job completed in 39 seconds with zero failed jobs. And I really don't care about the number of processed rows anymore, as the only measure relevant for billing now is runtime 😜

I also kept the verification close to the implementation: 25 focused tests cover SSH host-key parsing, disabled SSH-agent lookup, exact table scope, composite keys, cursor selection, cleanup, incremental state, and downstream SQL.

What I got for $0.65

I got a working dltHub job, focused tests, deployment configuration, a documented secret shape, and a better downstream behavior than the old Fivetran sync. I still had to make the data and security decisions. That part should remain human work.

But I spent an hour turning five pieces of manual context into a deployed pipeline instead of typing every tunnel option, test fixture, and metadata rule myself. That is a pretty good trade for $0.65 😎

The full pipeline:

"""Load 12 enabled Fivetran tables into Snowflake over SSH."""

import base64
import io
import logging
import select
import socket
import sys
import threading
import time

import dlt
import paramiko
from dlt.sources.sql_database import sql_database
import sqlalchemy as sa


logger = logging.getLogger(__name__)


SAFE_CURSOR_PRIORITY = (
    "changed",
    "content_translation_changed",
    "revision_timestamp",
)

TABLE_NAMES = (
    "media__field_image_caption",
    "media_revision__field_image_caption",
    "node__body",
    "node__field_shared_paragraphs",
    "node_field_data",
    "paragraph__field_pg_imgtxt_text",
    "paragraph__field_pg_media_text",
    "paragraph__field_pg_testimonial_testimonial",
    "paragraph__field_pg_text_text",
    "path_alias",
    "taxonomy_index",
    "taxonomy_term_field_data",
)


def parse_ssh_host_key(host_key):
    """Parse an OpenSSH ed25519 host-key line in memory."""
    key_type, encoded_key, *_ = host_key.split()
    if key_type != "ssh-ed25519":
        raise ValueError(f"unsupported SSH host-key type: {key_type}")
    return paramiko.Ed25519Key(data=base64.b64decode(encoded_key, validate=True))


class ParamikoTunnel:
    """Forward local TCP clients to a remote TCP service over SSH."""

    MAX_WORKERS = 64

    def __init__(self, hostname, username, private_key, host_key, target_address):
        self.hostname = hostname
        self.username = username
        self.private_key = private_key
        self.target_address = target_address
        self._client = paramiko.SSHClient()
        self._client.get_host_keys().add(hostname, host_key.get_name(), host_key)
        self._client.set_missing_host_key_policy(paramiko.RejectPolicy())
        self._listener = None
        self._accept_thread = None
        self._workers = set()
        self._connections = set()
        self._lock = threading.Lock()
        self._lifecycle_lock = threading.RLock()
        self._stop_event = threading.Event()
        self._stopping = False
        self._worker_slots = threading.BoundedSemaphore(self.MAX_WORKERS)
        self.local_bind_port = None

    def start(self):
        """Connect SSH and start a local listener on an ephemeral port."""
        with self._lifecycle_lock:
            if self._listener is not None:
                return

            try:
                self._client.connect(
                    hostname=self.hostname,
                    port=22,
                    username=self.username,
                    pkey=self.private_key,
                    allow_agent=False,
                    look_for_keys=False,
                )
            except Exception:
                self._client.close()
                raise
            listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            try:
                listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                listener.bind(("127.0.0.1", 0))
                listener.listen()
                listener.settimeout(0.2)
            except Exception:
                listener.close()
                self._client.close()
                raise

            with self._lock:
                self._stop_event.clear()
                self._stopping = False
                self._listener = listener
                self.local_bind_port = listener.getsockname()[1]
                self._accept_thread = threading.Thread(
                    target=self._accept_loop,
                    name="paramiko-tunnel-accept",
                    daemon=True,
                )
                self._accept_thread.start()

    def _accept_loop(self):
        listener = self._listener
        while not self._stop_event.is_set():
            try:
                client_socket, client_addr = listener.accept()
            except socket.timeout:
                continue
            except OSError:
                break

            with self._lock:
                stopping = getattr(self, "_stopping", False)
            if stopping or self._stop_event.is_set() or not self._worker_slots.acquire(
                blocking=False
            ):
                try:
                    client_socket.close()
                except Exception:
                    pass
                continue

            worker = threading.Thread(
                target=self._forward_connection,
                args=(client_socket, client_addr),
                name="paramiko-tunnel-forward",
                daemon=True,
            )
            with self._lock:
                if getattr(self, "_stopping", False) or self._stop_event.is_set():
                    self._worker_slots.release()
                    reject = True
                else:
                    reject = False
                    self._connections.add(client_socket)
                    self._workers.add(worker)
                    try:
                        worker.start()
                    except Exception:
                        self._connections.discard(client_socket)
                        self._workers.discard(worker)
                        self._worker_slots.release()
                        try:
                            client_socket.close()
                        except Exception:
                            pass
                        logger.exception("Paramiko tunnel worker failed to start")
            if reject:
                try:
                    client_socket.close()
                except Exception:
                    pass
                continue

    def _forward_connection(self, client_socket, client_addr):
        channel = None
        try:
            try:
                channel = self._client.get_transport().open_channel(
                    "direct-tcpip", self.target_address, client_addr
                )
                with self._lock:
                    self._connections.add(channel)
                self._relay(client_socket, channel, self._stop_event)
            except Exception:
                if not self._stop_event.is_set():
                    logger.exception("Paramiko tunnel worker failed")
        finally:
            for connection in (client_socket, channel):
                if connection is not None:
                    try:
                        connection.close()
                    except Exception:
                        pass
                    with self._lock:
                        self._connections.discard(connection)
            with self._lock:
                self._workers.discard(threading.current_thread())
            self._worker_slots.release()

    @staticmethod
    def _relay(client_socket, channel, stop_event=None):
        sources = {client_socket, channel}
        while sources:
            try:
                readable, _, _ = select.select(list(sources), [], [], 0.2)
            except Exception:
                if stop_event is None or not stop_event.is_set():
                    logger.exception("Paramiko tunnel relay select failed")
                return
            if not readable:
                continue
            for source in readable:
                try:
                    payload = source.recv(65536)
                except Exception:
                    if stop_event is None or not stop_event.is_set():
                        logger.exception("Paramiko tunnel relay receive failed")
                    return
                if not payload:
                    destination = channel if source is client_socket else client_socket
                    try:
                        if destination is channel:
                            destination.shutdown_write()
                        else:
                            destination.shutdown(socket.SHUT_WR)
                    except Exception:
                        pass
                    sources.remove(source)
                    continue
                destination = channel if source is client_socket else client_socket
                try:
                    destination.sendall(payload)
                except Exception:
                    if stop_event is None or not stop_event.is_set():
                        logger.exception("Paramiko tunnel relay send failed")
                    return

    def stop(self):
        """Stop accepting clients and close SSH, listener, and forwarded sockets."""
        with self._lifecycle_lock:
            with self._lock:
                self._stopping = True
                self._stop_event.set()
                listener = self._listener
                self._listener = None
            if listener is not None:
                try:
                    listener.close()
                except Exception:
                    pass

            with self._lock:
                connections = list(self._connections)
            for connection in connections:
                try:
                    connection.close()
                except Exception:
                    pass

            try:
                self._client.close()
            finally:
                deadline = time.monotonic() + 5
                while True:
                    with self._lock:
                        connections = list(self._connections)
                        workers = list(self._workers)
                    for connection in connections:
                        try:
                            connection.close()
                        except Exception:
                            pass
                    threads = [self._accept_thread, *workers]
                    for thread in threads:
                        if thread is not None and thread is not threading.current_thread():
                            thread.join(timeout=0.1)
                    if not any(thread is not None and thread.is_alive() for thread in threads):
                        break
                    if time.monotonic() >= deadline:
                        break
                self._accept_thread = None
                self.local_bind_port = None


def get_ssh_tunnel(ssh_credentials):
    """Build an SSH tunnel without opening the connection."""
    private_key = paramiko.Ed25519Key.from_private_key(
        io.StringIO(ssh_credentials["private_key"])
    )
    return ParamikoTunnel(
        hostname=ssh_credentials["server_ip_address"],
        username=ssh_credentials["username"],
        private_key=private_key,
        host_key=parse_ssh_host_key(ssh_credentials["host_key"]),
        target_address=("127.0.0.1", 3306),
    )


def create_mysql_engine(db_credentials, local_port):
    """Create a MySQL engine connected to the tunnel's local endpoint."""
    url = sa.URL.create(
        drivername=db_credentials.get("drivername", "mysql+pymysql"),
        username=db_credentials["username"],
        password=db_credentials["password"],
        host="127.0.0.1",
        port=local_port,
        database=db_credentials.get("database", "<hard_coded_db_name>"),
    )
    return sa.create_engine(url, poolclass=sa.pool.NullPool)


def create_source(engine):
    """Reflect the tables enabled for the conformably_fancies scope."""
    metadata = sa.MetaData()
    source = sql_database(
        engine,
        backend="pyarrow",
        metadata=metadata,
        table_names=list(TABLE_NAMES),
    )
    source.metadata = metadata
    return source


def choose_cursor(column_names):
    """Choose a known-safe incremental column."""
    columns = {name.lower(): name for name in column_names}
    for candidate in SAFE_CURSOR_PRIORITY:
        if candidate in columns:
            return columns[candidate]
    return None


def apply_resource_hints(resource, table):
    """Configure merge and incremental hints from reflected table metadata."""
    if table is None:
        resource.apply_hints(write_disposition="replace")
        return

    pk_columns = [column.name for column in table.primary_key.columns]
    cursor = choose_cursor([column.name for column in table.columns])
    if cursor is None or not pk_columns:
        resource.apply_hints(write_disposition="replace")
        return

    resource.apply_hints(
        write_disposition="merge",
        primary_key=pk_columns,
        incremental=dlt.sources.incremental(
            cursor, on_cursor_value_missing="include"
        ),
    )


def cleanup(engine, tunnel):
    """Dispose resources without masking a load exception."""
    active_exception = sys.exc_info()[0] is not None
    cleanup_error = None

    try:
        if engine is not None:
            engine.dispose()
    except Exception as error:
        cleanup_error = error
        if active_exception:
            print(f"warning: engine disposal failed: {error}")

    try:
        tunnel.stop()
    except Exception as error:
        if cleanup_error is None:
            cleanup_error = error
        if active_exception:
            print(f"warning: tunnel stop failed: {error}")

    if cleanup_error is not None and not active_exception:
        raise cleanup_error


def create_pipeline():
    """Create the production pipeline."""
    return dlt.pipeline(
        pipeline_name="webserver_cms",
        destination="snowflake",
        staging=dlt.destinations.filesystem(
            bucket_url="az://stage/webserver_cms"
        ),
        dataset_name="webserver_cms",
    )


def trigger_downstream_tasks(pipeline):
    """Refresh Cortex Search after a successful Snowflake load."""
    with pipeline.sql_client() as client:
        client.execute_sql(
            "execute task raw.webserver_cms.ta_refresh_cortex_search;"
        )
        print(
            "Executed task raw.webserver_cms.ta_refresh_cortex_search",
            flush=True,
        )


def load_webserver_cms():
    """Load the enabled Fivetran table scope."""
    ssh_credentials = dlt.secrets["sources.webserver_cms.ssh"]
    db_credentials = dlt.secrets["sources.webserver_cms.credentials"]
    tunnel = get_ssh_tunnel(ssh_credentials)
    engine = None

    try:
        tunnel.start()
        engine = create_mysql_engine(db_credentials, tunnel.local_bind_port)
        source = create_source(engine)
        metadata = getattr(source, "metadata", None)
        for table_name, resource in source.resources.items():
            print(f"discovered table: {table_name}")
            table = metadata.tables.get(table_name) if metadata is not None else None
            apply_resource_hints(resource, table)

        pipeline = create_pipeline()
        load_info = pipeline.run(source, loader_file_format="jsonl")
        print(load_info)
        trigger_downstream_tasks(pipeline)
        return load_info
    finally:
        cleanup(engine, tunnel)


if __name__ == "__main__":
    load_webserver_cms()