Day 17. Resources, config, IO managers and environments

⏱ 4h · learn 65m · lab 135m · ship 25m · review 15m🏷 hospitalityBefore: Day 16

By the end of today you can

Inject connections and settings with ConfigurableResource and EnvVar
Switch between local, test and prod without code changes
Use an IO manager to store asset outputs
Test assets with dg.materialize and fake resources

1. Why this matters

Hard-coded paths and connection strings are why pipelines can't be tested and can't move from a laptop to a server. Resources are Dagster's dependency injection: assets ask for "a warehouse" or "an SFTP client", and the environment decides which one.

2. Resources

# defs/resources.py
import dagster as dg
import fsspec, paramiko

class RawZone(dg.ConfigurableResource):
    url: str                                    # file://./data/raw  or  s3://raw
    def fs(self):
        return fsspec.url_to_fs(self.url)[0]

class SftpSource(dg.ConfigurableResource):
    host: str
    port: int = 22
    user: str
    password: str                               # comes from EnvVar, never hard-coded
    remote_dir: str = "/upload"
    def list_ready(self) -> list[str]:
        t = paramiko.Transport((self.host, self.port)); t.connect(username=self.user, password=self.password)
        with paramiko.SFTPClient.from_transport(t) as s:
            return sorted(f for f in s.listdir(self.remote_dir) if f.endswith(".txt"))

@dg.asset
def raw_pms_files(context, sftp: SftpSource, raw: RawZone) -> dg.MaterializeResult:
    ready = sftp.list_ready()
    ...

defs = dg.Definitions(
    assets=[...],
    resources={
        "sftp": SftpSource(host=dg.EnvVar("PMS_SFTP_HOST"), user=dg.EnvVar("PMS_SFTP_USER"),
                           password=dg.EnvVar("PMS_SFTP_PASSWORD")),
        "raw": RawZone(url=dg.EnvVar("P2D_RAW_URL")),
    },
)

dg.EnvVar is read at run time and not shown in the UI, unlike os.environ[...] evaluated at import.

3. Environments

Pick resources by deployment name:

import os
ENV = os.getenv("P2D_ENV", "local")
resources = {
    "local": {"raw": RawZone(url="file://./data/raw"), "warehouse": DuckDBWarehouse(path="data/wh.duckdb")},
    "prod":  {"raw": RawZone(url=dg.EnvVar("P2D_RAW_URL")), "warehouse": AzureSqlWarehouse(...)},
}[ENV]

Week 4 adds the warehouse resources. The asset code never changes.

4. IO managers

An IO manager decides where returned values go and how they're loaded downstream. dagster-polars and dagster-duckdb provide ready ones, so an asset can return df and the next asset receives df as an argument.

from dagster_duckdb_polars import DuckDBPolarsIOManager
defs = dg.Definitions(assets=[...], resources={"io_manager": DuckDBPolarsIOManager(database="data/wh.duckdb", schema="hotel")})

@dg.asset
def stg_bookings(raw_pms: pl.DataFrame) -> pl.DataFrame:   # loaded from DuckDB table hotel.raw_pms
    return pms.latest_version(raw_pms)                     # stored as hotel.stg_bookings

Maths and principle

Principle: dependency injection makes a function's environment an explicit argument, so it can be swapped.

Equation: instead of \(f(x)\) that secretly reads \(E\) (environment), write \(f(x, E)\). Tests evaluate \(f(x, E_{\text{fake}})\); prod evaluates \(f(x, E_{\text{prod}})\); same \(f\).

Analogy: in physics you make the boundary conditions explicit so the same equation solves different setups. Where it breaks: a resource with hidden global state (a cached connection reused across tests) leaks between environments.

Try it: an asset reads an env var at import time. Can a test change it per test? (Not reliably: the module was already imported. Use a resource or read inside the function.)

5. Testing assets

def test_stg_bookings_keeps_latest(tmp_path):
    raw = pl.DataFrame({"booking_id": [1, 1, 2], "status": ["confirmed", "cancelled", "confirmed"],
                        "file_received_at": ["2026-09-01", "2026-09-03", "2026-09-01"]})
    result = dg.materialize([stg_bookings], resources={"io_manager": dg.mem_io_manager}, ...)
    # or simply: assert stg_bookings(raw)["status"].to_list() == ["cancelled", "confirmed"]

Assets are plain functions; call them directly with fake inputs. Use dg.materialize when you want to test wiring and resources together.

The 5-year lens

Three settings files are a smell; one settings model with environment overrides is the pattern. And make "local" fully offline: DuckDB instead of Azure, a folder instead of SFTP, a fake API. If a new teammate (or interviewer) can git clone && just setup && dagster dev and see green assets in five minutes, your project is in the top few percent of portfolios.

Lab

Create resources RawZone, SftpSource, LocalFolderSource (same list_ready interface) and DuckDBWarehouse.
Refactor raw_pms_files to take a source resource. Local uses the folder; "docker" uses the atmoz/sftp container from Day 5.
Add P2D_ENV switching. Run both environments.
Try the DuckDB + Polars IO manager for stg_bookings and otb_by_stay_date. Query the tables in the DuckDB CLI.
Write 4 tests: one pure-function test per core function, one dg.materialize test of the hotel graph with the local folder fixture.

Ship

feat(dagster): resources, environments, IO manager, tests.