7 shared modules for 27 dlt pipelines

TL/DR: nothing in dlt or dltHub stops pipeline files from importing a shared helper module... After all, it's plain Python 😎 I run seven of them at my workspace root and they carry 27 pipelines.

My dltHub workspace has grown to roughly 70 jobs. Four of them load four PostgreSQL schemas out of the similar but separate Strapi CMS hosts (and no, that's not even all our CMS-backed websites 🙄) into Snowflake. The same structural functionality: reflection logic, batching, parquet staging, and a named downstream EXECUTE TASK call; but also, some differences: schema name, staging bucket, pipeline name, and (of course) which Snowflake task should actually be triggered afterwards (because they don't all have to trigger the same).

So, I pulled the load logic into a sahred _cms_common.py. And that made me realize, this approach might be useful on other pipelines, too. But since I couldn't find any reference or best-practice recommendation by dltHub on this, I hereby write it down myself 😜

What the docs say

The dltHub workspace docs describe a workspace as a collection of secret- and config-files, pyproject.toml, pipeline files, and an optional __deployment__.py manifest. Pipeline files are treated as opaque Python scripts: Whether those scripts import each other or a common helper is never discussed... so apparently, it's neither endorsed nor forbidden.

The dlt OSS docs go a bit further and actively recommend the pattern: The source page says a source is "most commonly" defined in a separate Python module from the actual pipeline. And the API tutorial has a dedicated (refactor) step where duplicated fetching code is extracted into a shared function:

You've noticed that there's a lot of code duplication in the get_issues and get_comments functions. We can reduce that by extracting the common fetching code into a separate function and using it in both resources.

OK, that's within one pipeline... but the rational doesn't change when the second consumer happens to be a different pipeline 😎

The actual pipeline

The shape I ended up with: the pipeline itself is pure configuration, while the helper does all the work. Here's one (abstracted version) of the CMS pipelines mentioned earlier in its entirety:

# dlt pipeline: CMS PostgreSQL for one example website → Snowflake.
# dltHub runtime schedule: "30 2 * * 1-5"

import dlt
from _cms_common import load_cms


def main():
    load_cms(
        pipeline_name="cms_example_website_to_snowflake",
        dataset_name="webserver_cms_example_website",
        credentials=dlt.secrets["sources.cms_example_website.credentials"],
        pg_schema="public",
        staging_bucket="az://stage/cms_example_website",
        downstream_tasks=[
            "raw.webserver_cms_example_website.ta_downstream"
        ],
    )


if __name__ == "__main__":
    main()

15 lines. The interesting part lives in _cms_common.py. There, the schema is reflected, (Strapi-CMS-) internal tables I don't care about are dropped, and small batches are loaded, because the managed Azure PG instance gets unhappy about too many concurrent connections:

def load_cms(
    pipeline_name: str,
    dataset_name: str,
    credentials: Any,
    pg_schema: str,
    staging_bucket: str,
    downstream_tasks: list[str] | None = None,
) -> None:
    """Generic CMS PostgreSQL → Snowflake loader."""
    pipeline = dlt.pipeline(
        pipeline_name=pipeline_name,
        destination="snowflake",
        staging=dlt.destinations.filesystem(bucket_url=staging_bucket),
        dataset_name=dataset_name,
    )

    source = sql_database(credentials=credentials, schema=pg_schema)
    content_tables = [
        name for name in source.resources.keys()
        if not name.startswith(EXCLUDED_PREFIXES)
    ]

    for i in range(0, len(content_tables), BATCH_SIZE):
        batch = content_tables[i : i + BATCH_SIZE]
        source = sql_database(
            credentials=credentials,
            schema=pg_schema,
            backend="pyarrow",
            engine_kwargs={"pool_size": 1, "max_overflow": 0},
        ).with_resources(*batch)
        pipeline.run(source, write_disposition="replace", loader_file_format="parquet")

    if downstream_tasks:
        with pipeline.sql_client() as client:
            for task in downstream_tasks:
                client.execute_sql(f"EXECUTE TASK {task}")

Four calling pipelines multiplied by about ~70 lines saved, minus ~70 lines of helper: about 210 lines gone. Cool. The interesting win came later, when at some point for some reason Azure PG started throttling and I had to reduce BATCH_SIZE from 4 to 2. One constant in just one (helper-) file, four pipelines fixed. Without the helper, would I have remembered all four files containing that knob? Maybe... 😅

Consistency is the icing on the cake: A reviewer opening cms_some_other_website_to_snowflake.py doesn't have to verify that it batches the same way as its three siblings. The import line already ensured it.

Seven modules, one concern each

I did not build one big utils.py. Instead, each module owns exactly one thing. And every helper-filename starts with an underscore to mark "infrastructure, not a pipeline" (naming conventions, yay 🥳):

About 450 lines of shared code total. Two of these are worth a closer look because they're not just deduplication.

_gcp_common.py exists because of one quirk: dlt.secrets reads the service-account JSON out of TOML, where the private_key newlines arrive escaped. They have to be un-escaped before google.oauth2 will accept them.

def get_credentials() -> service_account.Credentials:
    service_account_json = dlt.secrets["sources.gcp.service_account_json"]
    info = (service_account_json if isinstance(service_account_json, dict)
            else json.loads(service_account_json, strict=False))
    if isinstance(info, dict) and "private_key" in info:
        info["private_key"] = info["private_key"].replace("\\n", "\n")
    return service_account.Credentials.from_service_account_info(info)

If all BigQuery/GSC pipelines re-implemented that tiny transformation, one of them would eventually get it wrong... and the resulting 401 would be a fun afternoon. Centralising it makes the transformation explicit in one documented place.

_salesforce_common.py earns its keep differently: it holds an in-process auth cache with a 25-minute TTL. My Bulk API v2 pipelines instantiate one source per Salesforce object, up to 56 of them per run. Without the cache, that's up to 56 token roundtrips - with it, it's just one 😎

Import-time side effects

This approach has one (i.m.h.o. minor) disadvantage: _salesforce_common.py also sets a custom naming convention, because Snowflake uppercases unquoted identifiers and NPE03__DATE_ESTABLISHED__C needs to round-trip intact:

dlt.config["schema.naming"] = "_salesforce_common._SalesforceNamingConvention"
dlt.config["schema.allow_identifier_change_on_table_with_data"] = True

That has to happen at module import time, before any @dlt.source decorator is evaluated. And dlt.config is process-global, which means: if __deployment__.py imported this module at the top of the file, every other pipeline sharing that Python process would silently inherit the Salesforce naming convention.

The fix is to import the helper lazily, inside the job trampoline:

@job(trigger=schedule("30 2 * * *"), execute={"timeout": "30m"})
def salesforce_to_snowflake():
    import salesforce_to_snowflake as _mod
    _mod.main()

And another gotcha: dlt persists the naming convention as a string reference in the saved schema. So _salesforce_common can never be renamed without breaking the next run, and any tool that calls dlt.attach() has to be able to import it. The dltHub-workspace MCP server (which is such a tool) fails with UnknownNamingModule if it launches from a cwd where that file isn't on sys.path. Pipeline runs are unaffected, but agent introspection breaks here.

When I don't share

I only extract pieces of code from a pipeline when two or more currently existing pipelines substantially utilize the same logic. YAGNI applies here: a shared function signature becomes a small public API only when a second caller shows up.

So, these examples stayed inline on purpose: A sha2-dedupe and lateral-flatten logic in one pipeline, a byte-for-byte CSV formatting helpers to produce CSVs for an external consumer in a different pipeline, and dlt.config["schema.naming"] = "sql_ci_v1" in another pipeline where a different naming would impact downstream processes.

However, if a shared function starts growing branches (stuff like if pipeline_name == "cms_example_website" ), the abstraction is leaking and the pipelines should go back to being separate.

One gap I promise to get to some day: The shared modules have no isolated tests, so a bug in one helper surfaces in whichever consumer happens to run first.

And that's it: seven small helper files in a dlt workspace... Python doing what Python does 😎