> ## Content Index > Fetch the complete content index at: https://www.sfrt.io/llms.txt > Use this file to discover other available public pages before exploring further. # 7 shared modules for 27 dlt pipelines - URL: https://www.sfrt.io/7-shared-modules-for-27-dlt-pipelines/ - Published: 2026-08-03T05:40:17.000Z - Updated: 2026-08-03T05:40:16.000Z - Description: Re-using shared patterns in a dlt workspace - Author: Martin Seifert - Tags: Infrastructure 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](https://strapi.io/?ref=sfrt.io) 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](https://dlthub.com/docs/tutorial/load-data-from-an-api?ref=sfrt.io#dynamic-resources) 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 😎 One [documented](https://dlthub.com/docs/hub/pipeline-operations/job-configuration?ref=sfrt.io#job-configuration-via-toml) aspect matters here: job config sections default to the containing module name (`[jobs.my_pipeline]`). A helper module is not a job, so it never gets a config section of its own. The means, anything a shared helper reads has be referenced explicitly. ## 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: ```python # 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: ```python 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 🥳): | Module | LOC | Adopters | Does | | ------------------------- | --- | -------- | ------------------------------------------------------------------------------------------------------------------- | | \_azure\_table\_common.py | 50 | 3 | Azure Table Storage connection string + transforming UTC timestamps to Zurich timezone | | \_cms\_common.py | 70 | 4 | The generic CMS PostgreSQL → Snowflake loader | | \_gcp\_common.py | 25 | 4 | Authenticate with my GCP project (to access BigQuery or GSC or whatever) from service-account JSON | | \_salesforce\_common.py | 150 | 5 | OAuth2 client, auth cache, naming convention, ... | | \_send\_email\_common.py | 50 | 3 | POST to a shared Power Automate flow sending an email | | \_sftp\_common.py | 40 | 5 | paramiko SFTP session (I keep being surprised how often I use this!) | | \_snowflake\_common.py | 65 | 3 | SQLAlchemy engine (not ADBC because it only reads exactly 1 table/view) + read\_query() to load data from Snowflake | 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. ```python 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](https://developer.salesforce.com/docs/atlas.en-us.api%5Fasynch.meta/api%5Fasynch/bulk%5Fapi%5F2%5F0.htm?ref=sfrt.io) 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: ```python 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: ```python @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 😎