Simple ETL Configuration with Pydantic Settings

Small ETL projects do not need a complicated configuration layer. Most of the time I want three things: read a few values from .env, derive predictable file paths from the project root, and make sure the expected folders exist before the pipeline touches them.

For that shape, I like one Settings class with typed environment values and computed path fields. The environment controls file names and runtime knobs; the Python code owns path construction.

Project Shape

etl/
├── .env
├── data/          # raw/ and out/ created at startup
└── src/etl/
    ├── config.py
    └── pipeline.py

ETL_BATCH_SIZE=5000
ETL_INPUT_FILE=customers.csv
ETL_OUTPUT_FILE=customers_clean.parquet

Settings

from functools import cached_property
from pathlib import Path

from pydantic import Field, computed_field
from pydantic_settings import BaseSettings, SettingsConfigDict


ROOT = Path(__file__).resolve().parents[2]


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=ROOT / ".env",
        env_prefix="ETL_",
        extra="ignore",
    )

    batch_size: int = Field(default=1000, ge=1)
    input_file: str = "customers.csv"
    output_file: str = "customers_clean.parquet"

    @computed_field
    @cached_property
    def raw_dir(self) -> Path:
        return ROOT / "data" / "raw"

    @computed_field
    @cached_property
    def out_dir(self) -> Path:
        return ROOT / "data" / "out"

    @computed_field
    @cached_property
    def input_path(self) -> Path:
        return self.raw_dir / self.input_file

    @computed_field
    @cached_property
    def output_path(self) -> Path:
        return self.out_dir / self.output_file


settings = Settings()

print(settings.batch_size)
print(settings.input_path)
print(settings.output_path)

# --------------------
5000
/etl/data/raw/customers.csv
/etl/data/out/customers_clean.parquet

The useful split is small but important: BaseSettings reads and validates the values that can change between environments, while the computed fields keep all path construction in one place.

Why @computed_field on top of @cached_property?

@cached_property alone already does the runtime job: each path is computed lazily, once per instance, and attribute access works exactly the same. So why the extra decorator?

Because a plain cached_property is invisible to Pydantic. It is just a Python attribute — model_dump() does not know it exists. @computed_field registers the property as part of the model, and that changes what the settings object is: not three raw env values with some helper attributes bolted on, but the full effective configuration, resolved paths included.

That matters the moment something goes wrong. When a run fails, the first question is "what configuration did it actually use?" Log settings.model_dump() at startup and, with computed fields, the answer is complete — including exactly where the pipeline looked for input and where it wrote output. Without them, the dump says input_file='customers.csv' and stays silent about which directory that resolved to, which is the one piece of information you actually need. The same introspection also powers the directory setup below.

The caching comes with one caveat: do not mutate settings.input_file later and expect settings.input_path to follow. Treat the settings object as a read-only snapshot and create a new Settings() if the configuration changes.

Self-Documenting Directory Setup

There is one naming rule in the class doing quiet work: every field that represents a directory ends in _dir, and every file location derived from one ends in _path. That convention plus model_dump() gives a bootstrap function that never needs to be told what to create:

def ensure_dirs(settings: Settings) -> list[Path]:
    dirs = [
        value
        for name, value in settings.model_dump().items()
        if name.endswith("_dir")
    ]
    for path in dirs:
        path.mkdir(parents=True, exist_ok=True)
    return dirs


for path in ensure_dirs(settings):
    print(path)

# --------------------
/etl/data/raw
/etl/data/out

Why this earns its place:

• It kills a real failure mode: the drifting bootstrap list. The naive version enumerates directories by hand — for path in (settings.raw_dir, settings.out_dir): ... — and that list must be edited every time the configuration grows. Forget it once, and the pipeline crashes on its first write. With discovery via model_dump(), adding a staging_dir field is the whole change; creation follows automatically, so configuration and bootstrap code cannot drift apart.
• The convention documents intent. Anyone scanning the class knows that every *_dir is a folder the pipeline guarantees at startup and every *_path is a file inside one. The suffix is not decoration; it is a small contract that code relies on.
• It only works because of @computed_field. model_dump() sees registered fields only — with bare cached_property the dump would contain just the env values and ensure_dirs would silently create nothing. This is the concrete payoff for the extra decorator.

Since mkdir(parents=True, exist_ok=True) is idempotent, calling it at every startup is safe. And note what did not change: creating Settings() still only reads configuration. Touching the filesystem stays in an explicit function you call at the edge of the pipeline.

In the Pipeline

from etl.config import Settings, ensure_dirs

settings = Settings()
ensure_dirs(settings)

df = read_customers(settings.input_path, batch_size=settings.batch_size)
clean = clean_customers(df)
write_customers(clean, settings.output_path)

print(clean.head(3).to_string(index=False))

# --------------------
 customer country  revenue
Company A      AT     1200
Company B      DE      940
Company C      AT      650

That is enough for most small ETL scripts: typed env values, visible derived paths, directories that create themselves from a naming convention, and no hidden configuration layer doing too much behind the scenes.