Build a Weather Data Pipeline with Dagster

Orchestrate a live weather data pipeline from API to PostgreSQL with Dagster.

Introduction

30 Second Summary

Weather apps update their forecasts automatically every few hours. Behind the scenes, a data pipeline pulls fresh numbers from an API, cleans them up, and stores them for display.

In this project, you will build your own weather data pipeline using Dagster. Your pipeline will fetch live forecast data for Indonesian cities, load it into PostgreSQL, and transform it into daily summaries with automated quality checks and a daily schedule.

What You'll Build

Click one button in the Dagster UI and watch live weather data flow from an API through your database into clean daily summaries, with every step visible in a real-time lineage graph.

By the end of this project, you'll have:

A one-click pipeline that pulls live weather forecasts for three Indonesian cities, loads them into a database, and produces clean daily summaries you can query and preview right in the Dagster UI.
Automated data quality checks that validate every run, catching null values, out-of-range temperatures, and missing cities before bad data spreads.
A daily schedule that runs the full pipeline at 6 AM without you lifting a finger, visible in the Dagster Automations tab.
Secret Mission: Build a city weather ranking that identifies the hottest, wettest, and windiest city each day.

Are there any prerequisites?

This project assumes you are comfortable with Python and SQL. You should also have Docker installed and be familiar with using the terminal.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Docker, PostgreSQL, and Dagster

Building a data pipeline needs two things: a database to store data and an orchestrator to manage the pipeline.

In this step, you will set up PostgreSQL as your database using Docker, and install Dagster as your pipeline orchestrator. Dagster replaces manual "run this script, then that script" workflows with a visual, dependency-aware system.

In this step, get ready to:

Install uv and scaffold a Dagster project.
Start a PostgreSQL database using Docker Compose.
Install Python dependencies and confirm the Dagster UI is running.

Install uv and scaffold the Dagster project

Dagster uses uv as its recommended Python package manager. You will use uv to scaffold and manage the entire Dagster project.

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
Check if uv is already installed by running:

uv --version

✔️ I see a version number

You already have uv installed. Move on to the next instruction below.

ⓧ Command not found

You need to install uv. Run the install command for your operating system:

macOS

Install uv by running:

curl -LsSf https://astral.sh/uv/install.sh | sh

Windows

Install uv by running:

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

Close and reopen your terminal so the uv command is available on your PATH.
Verify the install by running uv --version again. You should see a version number printed.

Still seeing command not found?

Make sure you closed and reopened your terminal after installing. The installer adds uv to your PATH, but that change only takes effect in a new terminal session.

help me troubleshoot uv installation issues on my machine

Move to your Desktop by running this command:

cd ~/Desktop

Scaffold a new Dagster project by running this command:

uvx create-dagster@latest project weather_pipeline

What does this command do?

The uvx command runs create-dagster without installing it globally. It scaffolds a new Dagster project inside a weather_pipeline folder with all the standard files and directories Dagster needs.

The command will prompt you to run uv sync. This sets up a virtual environment and installs Dagster's core dependencies.

When you see the prompt asking to run uv sync, type y and press Enter.
Wait for the sync to finish. This may take a minute as it downloads Dagster and its dependencies.
Move into the project folder by running:

cd weather_pipeline

Activate the virtual environment that uv sync created:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

Verify Dagster is installed by running:

dg --version

You should see a version number printed. This confirms Dagster and the dg CLI are ready.

Seeing command not found for dg?

Make sure you activated the virtual environment first. The dg command is only available inside the project's .venv.

help me troubleshoot why the dg command is not found after scaffolding a Dagster project

Start PostgreSQL with Docker Compose

Your pipeline needs a database to store the weather data it extracts. You will run PostgreSQL as a Docker container so there is nothing to install on your machine.

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type VS Code and press Enter to open it.
Click File in the top menu bar.
Click Open Folder.
Select the weather_pipeline folder on your Desktop, then click Open.
Create a new file called docker-compose.yml in the root of the weather_pipeline folder (alongside pyproject.toml). Copy the following code into it:

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: dagster_user
      POSTGRES_PASSWORD: dagster_pass
      POSTGRES_DB: weather_db
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

What does this file do?

image: postgres:16 tells Docker to pull the PostgreSQL 16 image.
The environment block creates a database called weather_db with a user dagster_user and password dagster_pass.
Port 5432 is mapped to your machine so Python can connect to the database at localhost:5432.
The pgdata volume keeps your data safe even if the container restarts.

Save docker-compose.yml.
Open a terminal in VS Code by pressing Cmd+` (macOS) or Ctrl+` (Windows).
Start the PostgreSQL container by running:

docker compose up -d

What does the -d flag do?

The -d flag runs the container in detached mode. This means it runs in the background so your terminal stays free for other commands.

Verify that PostgreSQL is running by running:

docker compose ps

You should see a postgres container listed with a status of running.

Container not running?

Make sure Docker Desktop is running before you run docker compose up -d. Open Docker Desktop and wait for it to finish starting.

If port 5432 is already in use, another PostgreSQL instance may be running on your machine. Stop it first, then try again.

help me troubleshoot why my Docker Compose PostgreSQL container is not starting

✔️ Awesome, I've got everything!

Great. Double check you saved docker-compose.yml and that docker compose ps shows the container running.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml. Make sure your file matches exactly:

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: dagster_user
      POSTGRES_PASSWORD: dagster_pass
      POSTGRES_DB: weather_db
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Install dependencies and start the Dagster UI

Your Dagster project needs a few Python libraries to fetch weather data, process it, and connect to PostgreSQL.

Back in the VS Code terminal, make sure your virtual environment is active (you should see (.venv) in your prompt). If not, activate it again using the platform-specific command from earlier.
Install the project dependencies by running:

uv add requests pandas sqlalchemy psycopg2-binary

What are these packages?

requests makes HTTP calls to the weather API.
pandas handles data manipulation and transformation.
sqlalchemy provides a Python interface for reading from and writing to databases.
psycopg2-binary is the PostgreSQL driver that sqlalchemy uses under the hood.

You should see output showing each package being resolved and installed.

Before you start the Dagster UI, what do you expect to see when it loads? There are no assets defined yet.

Start the Dagster development server by running:

dg dev

What does dg dev do?

This command starts a local Dagster webserver and opens the UI at http://localhost:3000. It also watches for file changes so it reloads automatically as you add code.

The dg dev process takes over your terminal while it runs. You will need to open a new terminal tab or window for future commands.

Open your browser and go to http://localhost:3000.

The Dagster UI loads showing an empty asset graph. This is expected because you have not defined any assets yet.

Dagster UI not loading?

Make sure dg dev is still running in your terminal. If it exited with an error, check that your virtual environment is activated.

If port 3000 is in use by another application, dg dev may fail to start. Close whatever is using that port and try again.

help me troubleshoot why the Dagster UI is not loading at localhost:3000

Your database is running and your orchestrator is live. Next up, you will write your first Dagster asset that pulls live weather data from the Open-Meteo API and loads it into PostgreSQL.

Build Your First Extract Asset

Your pipeline needs data. Right now the Dagster UI shows an empty asset graph with nothing to run.

In this step, you will write a Python function that fetches live weather data from the Open-Meteo API and loads it into PostgreSQL. This is the extract and load portion of your pipeline.

In this step, get ready to:

Create the raw_weather_forecast asset that calls the Open-Meteo API for three Indonesian cities.
Materialize the asset in the Dagster UI and inspect its metadata.

Write the raw weather forecast asset

In Dagster, every piece of data your pipeline produces is an asset. An asset is a Python function decorated with @dg.asset that produces a dataset Dagster can track, visualize, and schedule.

In VS Code, create a new file called assets.py inside the src/weather_pipeline/defs/ folder.
Start the file with the imports, database connection string, and the opening of the asset function by copying this code into assets.py:

import requests
import pandas as pd
from sqlalchemy import create_engine
import dagster as dg

DATABASE_URL = "postgresql://dagster_user:dagster_pass@localhost:5432/weather_db"


@dg.asset(group_name="extract")
def raw_weather_forecast(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    """Fetch 7-day hourly weather forecast for Indonesian cities from Open-Meteo API."""
    cities = [
        {"name": "Jakarta", "latitude": -6.2088, "longitude": 106.8456},
        {"name": "Surabaya", "latitude": -7.2575, "longitude": 112.7521},
        {"name": "Bandung", "latitude": -6.9175, "longitude": 107.6191},
    ]

What does this code do?

DATABASE_URL stores the connection string for the PostgreSQL container you started in Step 1. It points to the weather_db database using the credentials from your docker-compose.yml.
@dg.asset(group_name="extract") registers this function as a Dagster asset in the "extract" group. Dagster will display it in the asset graph.
The cities list defines the three Indonesian cities to fetch weather data for, each with GPS coordinates the Open-Meteo API needs.

The function continues in the next chunk. Do not save the file yet.

Add the API call loop below the cities list, still inside the raw_weather_forecast function:

    all_data = []
    for city in cities:
        response = requests.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": city["latitude"],
                "longitude": city["longitude"],
                "hourly": "temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m",
                "timezone": "auto",
                "forecast_days": 7,
            },
        )
        response.raise_for_status()
        data = response.json()

        hourly = data["hourly"]
        df = pd.DataFrame(
            {
                "city": city["name"],
                "latitude": city["latitude"],
                "longitude": city["longitude"],
                "timestamp": hourly["time"],
                "temperature_c": hourly["temperature_2m"],
                "humidity_pct": hourly["relative_humidity_2m"],
                "precipitation_mm": hourly["precipitation"],
                "wind_speed_kmh": hourly["wind_speed_10m"],
            }
        )
        all_data.append(df)

What does this loop do?

The for loop calls the Open-Meteo API once per city, requesting 7 days of hourly data for temperature, humidity, precipitation, and wind speed.
response.raise_for_status() stops the pipeline if any API call fails, so your pipeline halts instead of silently loading bad data.
Each city's response is reshaped into a pandas DataFrame with descriptive column names like temperature_c and precipitation_mm, then appended to all_data.

Why is no API key needed?

Open-Meteo is free for non-commercial use with up to 10,000 daily API calls. No sign-up, no API key, no credit card. You call the endpoint and get JSON back.

One more chunk to go. Do not save the file yet.

Add the final section below the for loop (back at the top indentation level of the function, not inside the loop) to combine, load, and return the data:

    result = pd.concat(all_data, ignore_index=True)

    engine = create_engine(DATABASE_URL)
    result.to_sql("raw_weather_forecast", engine, if_exists="replace", index=False)
    engine.dispose()

    context.log.info(f"Loaded {len(result)} rows for {len(cities)} cities")

    return dg.MaterializeResult(
        metadata={
            "num_rows": len(result),
            "num_cities": len(cities),
            "preview": dg.MetadataValue.md(
                result.head(10).to_markdown(index=False)
            ),
        }
    )

What does this section do?

pd.concat merges all three city DataFrames into a single DataFrame.
to_sql writes the combined data to a PostgreSQL table called raw_weather_forecast, replacing any previous data each time the asset runs.
MaterializeResult returns metadata that Dagster displays in the UI: the row count, city count, and a markdown preview of the first 10 rows.

Save assets.py.

Seeing a red squiggly line or import error?

Make sure the indentation is consistent. Everything inside the raw_weather_forecast function must be indented by 4 spaces. The for loop body must be indented by 8 spaces.
Check that the final result = pd.concat(...) block is at the same indentation level as all_data = [] (4 spaces), not inside the for loop.
Confirm that your virtual environment is active. VS Code may highlight imports as missing if the wrong Python interpreter is selected.

Why is my assets.py showing errors in VS Code?

✔️ Awesome, I've got everything!

Great. Double check you have saved assets.py before moving on.

ⓧ I'd like to double check the full code

Your assets.py file is made up of the three chunks you just added. Scroll up and confirm each one is present, in this order:

Chunk 1: Imports (requests, pandas, sqlalchemy, dagster), the DATABASE_URL constant, the @dg.asset decorator, function signature, and the cities list.
Chunk 2: The for city in cities loop that calls the Open-Meteo API and builds a DataFrame per city.
Chunk 3: pd.concat, to_sql, logging, and the MaterializeResult return.

Make sure the entire function body is indented under def raw_weather_forecast and that the concat/load block is outside the for loop.

Materialize the asset in the Dagster UI

Your asset code is saved. Now you will trigger Dagster to actually run the function and load data into PostgreSQL.

Switch back to your browser with the Dagster UI at http://localhost:3000.
Click **Assets** in the top navigation bar.
Click Reload definitions to tell Dagster to pick up your new asset file.

You should see raw_weather_forecast appear in the asset graph. The asset is defined but has not run yet.

Don't see the asset after reloading?

Check the terminal running dg dev for error messages. A syntax error in assets.py will prevent Dagster from loading your definitions.
Make sure the file is saved at src/weather_pipeline/defs/assets.py. Dagster auto-discovers files inside the defs/ folder.

My asset is not showing up in the Dagster UI after reloading definitions.

Before you materialize, think about this: the API returns hourly data for 7 days across 3 cities. How many rows of data do you expect to see?

Click Materialize all in the top right of the asset graph.

The asset turns green. The answer to the row count question: 3 cities x 7 days x 24 hours = 504 rows.

Click on the raw_weather_forecast asset in the graph to open its detail panel.
Look at the metadata. You should see num_rows: 504, num_cities: 3, and a preview table showing the first 10 rows of weather data with columns like temperature_c and wind_speed_kmh.

What just happened?

Dagster ran your raw_weather_forecast function, which called the Open-Meteo API three times (once per city), combined the results, and wrote 504 rows of hourly weather data into the raw_weather_forecast table in PostgreSQL.

The metadata you see in the UI is what MaterializeResult returned. Dagster stores this alongside the run so you can inspect any past run's output without re-running the pipeline.

Materialization failed or asset turned red?

Click the failed run in the Dagster UI to read the error message. A ConnectionRefusedError means your PostgreSQL container is not running. Return to your terminal and run docker compose up -d from the project folder.
A requests.exceptions.ConnectionError means the Open-Meteo API is unreachable. Check your internet connection and try materializing again.
If you see an IndentationError, go back to assets.py and make sure the concat/load block (chunk 3) is at the same indentation level as all_data = [] (4 spaces), not inside the for loop.

My raw_weather_forecast materialization failed in Dagster. Help me debug.

Your first asset is live and pulling real weather data into PostgreSQL. Next up, you will transform that raw hourly data into daily summaries and connect the two assets with a dependency.

Build a Transform Asset with Dependencies

In the last step, you pulled 504 rows of live hourly weather data from the Open-Meteo API and loaded it into PostgreSQL. Your extract asset is green in the Dagster UI.

Raw hourly data is not useful for analysis. You need daily summaries per city: average temperature, total precipitation, max wind speed. The transform asset you build in this step must depend on the extract so Dagster knows to run them in the right order.

In this step, get ready to:

Add the daily_weather_summary asset with a dependency on raw_weather_forecast.
Materialize both assets and observe the lineage graph connecting them.

Add the daily weather summary asset

Your raw data has 24 rows per city per day. A daily summary collapses those 24 hourly readings into one row with averages, totals, and extremes. In Dagster, you express this as a second asset that depends on the first.

In VS Code, open src/weather_pipeline/defs/assets.py.
Scroll to the bottom of the file, after the closing line of the raw_weather_forecast function.
Add the first part of the daily_weather_summary asset by pasting this code below the existing function:

@dg.asset(deps=[raw_weather_forecast], group_name="transform")
def daily_weather_summary(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    """Aggregate hourly weather data into daily summaries per city."""
    # Read raw hourly data from PostgreSQL
    engine = create_engine(DATABASE_URL)
    raw_df = pd.read_sql("SELECT * FROM raw_weather_forecast", engine)

    # Extract just the date from each timestamp
    raw_df["date"] = pd.to_datetime(raw_df["timestamp"]).dt.date

    # Group by city and date, then calculate daily aggregates
    daily = (
        raw_df.groupby(["city", "date", "latitude", "longitude"])
        .agg(
            avg_temperature_c=("temperature_c", "mean"),
            min_temperature_c=("temperature_c", "min"),
            max_temperature_c=("temperature_c", "max"),
            avg_humidity_pct=("humidity_pct", "mean"),
            total_precipitation_mm=("precipitation_mm", "sum"),
            max_wind_speed_kmh=("wind_speed_kmh", "max"),
        )
        .reset_index()
    )

What does this code do?

deps=[raw_weather_forecast] tells Dagster that this asset depends on the extract asset. Dagster will always run the extract first.
pd.read_sql reads the raw hourly data straight from PostgreSQL into a pandas DataFrame.
.dt.date strips the time from each timestamp so you can group by calendar date.
.groupby().agg() collapses 24 hourly rows per city per day into one row with mean, min, max, and sum values.

Save assets.py.
Switch to the Dagster UI at http://localhost:3000 and refresh the page. You should see daily_weather_summary appear as a new asset in the graph, connected to raw_weather_forecast by an arrow.

Don't see the new asset in the graph?

Make sure you saved assets.py. The Dagster dev server reloads automatically on save.
Check the terminal where dg dev is running for any Python errors. A syntax error in the new function will prevent the module from loading.
Confirm the new function is pasted after the raw_weather_forecast function, not inside it.

Still stuck? Ask for help.

The asset appears in the graph, but the function is not finished yet. It reads and aggregates data, but it does not write the results or return metadata to Dagster.

Complete the daily_weather_summary function by adding this code directly below the .reset_index() line, inside the function (indented one level):

    # Round decimal values for cleaner output
    daily["avg_temperature_c"] = daily["avg_temperature_c"].round(1)
    daily["avg_humidity_pct"] = daily["avg_humidity_pct"].round(1)
    daily["total_precipitation_mm"] = daily["total_precipitation_mm"].round(1)
    daily["max_wind_speed_kmh"] = daily["max_wind_speed_kmh"].round(1)

    # Write daily summaries to PostgreSQL
    daily.to_sql("daily_weather_summary", engine, if_exists="replace", index=False)
    engine.dispose()

    context.log.info(f"Created {len(daily)} daily summaries")

    # Return metadata for the Dagster UI
    return dg.MaterializeResult(
        metadata={
            "num_rows": len(daily),
            "preview": dg.MetadataValue.md(
                daily.head(10).to_markdown(index=False)
            ),
        }
    )

What does this code do?

.round(1) rounds each aggregated column to one decimal place for cleaner output.
.to_sql("daily_weather_summary", ...) writes the daily summary DataFrame into a new PostgreSQL table.
MaterializeResult sends metadata back to the Dagster UI: a row count and a markdown preview of the first 10 rows.

Save assets.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved assets.py before moving on.

ⓧ I'd like to double check the full code

Your assets.py file should contain, in order:

The import lines for requests, pandas, create_engine, and dagster from the previous step.
The DATABASE_URL constant from the previous step.
The full raw_weather_forecast function from the previous step.
The complete daily_weather_summary function you just built in the two chunks above (aggregation, then rounding and metadata).

Scroll up and confirm each section is present in your file, in this order.

Materialize both assets and observe the lineage graph

Your transform asset is defined, but it has never run. Time to materialize both assets and see Dagster execute them in dependency order.

Before you click the button, think about this: what order do you expect Dagster to run the two assets in, and why?

In the Dagster UI at http://localhost:3000, click Materialize All in the top right corner of the asset graph.
Watch the run progress. Dagster materializes raw_weather_forecast first, then daily_weather_summary. Both assets turn green.

Why did the extract run first?

You declared deps=[raw_weather_forecast] on the transform asset. Dagster reads that dependency and automatically runs the extract before the transform. You never have to manage execution order yourself.

Click on the daily_weather_summary asset in the graph to open its detail panel.
Check the metadata section. You should see a row count and a preview table showing daily aggregated data with columns like avg_temperature_c, total_precipitation_mm, and max_wind_speed_kmh.
Look at the lineage graph. An arrow flows from raw_weather_forecast into daily_weather_summary, showing the dependency you defined.

Materialization failed?

Click the failed run in the Dagster UI to see the error log. A common cause is a missing indentation on the second chunk. Every line you added must be indented inside the daily_weather_summary function.
If you see a database connection error, confirm your Docker PostgreSQL container is still running by checking docker compose ps in a separate terminal.

Need help debugging?

Both assets are green. The lineage graph shows data flowing from extract to transform. The metadata panel shows aggregated numbers for each city and day.

But look closely at those numbers. How do you know they are correct? If the API returned garbage temperatures, your pipeline would aggregate the garbage into clean-looking daily summaries and report success. There is no validation.

Why is silent success a problem?

In production, silent data corruption is worse than a crash. A crash stops the pipeline and alerts someone. Silent corruption flows downstream into dashboards, reports, and decisions without anyone knowing the numbers are wrong.

Your pipeline needs automated guardrails that check the output after every run. Dagster has a feature for exactly this: asset checks.

Your two-asset pipeline is running and producing daily weather summaries. Next up, you will add asset checks that automatically validate the data after every run, so your pipeline catches problems instead of hiding them.

Add Asset Checks for Data Quality

Your pipeline now has two assets connected by a dependency, and the lineage graph proves they run in the right order.

There is a gap. Your pipeline reports success even if the data is wrong. If the API returned garbage values, your daily summaries would silently produce incorrect results with no warning. In production, silent data corruption is worse than a crash.

In this step, get ready to:

Write three asset checks that validate your daily summaries after every pipeline run.
Materialize the pipeline and confirm all checks pass with green checkmarks in the Dagster UI.

Create asset checks for your daily summaries

Asset checks are functions that run automatically after an asset materializes. They validate the output and report pass or fail, making data problems visible instead of silent.

In VS Code, create a new file called checks.py inside the src/weather_pipeline/defs/ folder.
Add the imports and first asset check by copying the following code into checks.py:

import pandas as pd
from sqlalchemy import create_engine
import dagster as dg
from .assets import daily_weather_summary, DATABASE_URL


# Check 1: no null temperature values
@dg.asset_check(asset=daily_weather_summary, description="Verify no null temperature values in daily summaries")
def no_null_temperatures() -> dg.AssetCheckResult:
    engine = create_engine(DATABASE_URL)
    df = pd.read_sql("SELECT * FROM daily_weather_summary", engine)
    engine.dispose()

    null_count = int(df["avg_temperature_c"].isna().sum())
    return dg.AssetCheckResult(
        passed=null_count == 0,
        metadata={"null_temperature_count": null_count},
    )

What does this code do?

The imports bring in daily_weather_summary and DATABASE_URL from your assets.py file. This connects each check to the asset it validates.

The @dg.asset_check decorator works like @dg.asset, but instead of producing data, it validates data. The asset parameter tells Dagster which asset this check belongs to.

The function reads the daily summary table, counts how many null temperature values exist, and returns a pass or fail result. The metadata dictionary attaches the actual null count to the check result so you can inspect it in the UI.

Save checks.py.
Switch to the Dagster UI at http://localhost:3000.
Click Reload definitions in the top right of the asset graph to pick up your new file.
You should see a small check indicator appear on the daily_weather_summary asset in the graph. That means Dagster found your check.

Don't see the check indicator?

Make sure your file is saved as checks.py inside src/weather_pipeline/defs/, not in the project root.
Check the terminal running dg dev for import errors. A typo in the import line will prevent the file from loading.
Make sure you clicked Reload definitions after saving. Dagster does not auto-reload code changes.

Having trouble? Why isn't my asset check showing up in the Dagster UI?

Now add the second check below no_null_temperatures in checks.py:

# Check 2: temperatures within reasonable range for Indonesia
@dg.asset_check(asset=daily_weather_summary, description="Verify temperatures are within reasonable range for Indonesia")
def temperature_in_range() -> dg.AssetCheckResult:
    engine = create_engine(DATABASE_URL)
    df = pd.read_sql("SELECT * FROM daily_weather_summary", engine)
    engine.dispose()

    out_of_range = df[
        (df["avg_temperature_c"] < -10) | (df["avg_temperature_c"] > 50)
    ]
    return dg.AssetCheckResult(
        passed=len(out_of_range) == 0,
        metadata={"out_of_range_count": len(out_of_range)},
    )

What does this check validate?

This check filters for any rows where the average temperature falls below -10 or above 50 degrees Celsius. Indonesian cities should never hit those extremes. If the API returned corrupted data (like 999 degrees), this check would catch it and fail.

Add the third and final check below temperature_in_range in checks.py:

# Check 3: all expected cities are present
@dg.asset_check(asset=daily_weather_summary, description="Verify all 3 cities have data in the daily summary")
def all_cities_present() -> dg.AssetCheckResult:
    engine = create_engine(DATABASE_URL)
    df = pd.read_sql("SELECT * FROM daily_weather_summary", engine)
    engine.dispose()

    city_count = int(df["city"].nunique())
    return dg.AssetCheckResult(
        passed=city_count == 3,
        metadata={"city_count": city_count},
    )

What does this check validate?

This check counts the distinct cities in the summary table and verifies there are exactly 3 (Jakarta, Surabaya, and Bandung). If the API call for one city failed silently, this check would catch the missing data.

Save checks.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved checks.py with all three checks before moving on.

ⓧ I'd like to double check the full code

Your checks.py file should contain these sections in order. Scroll up and confirm each one is present:

Imports: pandas, sqlalchemy, dagster, and the relative import from .assets.
Check 1: no_null_temperatures function decorated with @dg.asset_check.
Check 2: temperature_in_range function that filters for values outside -10 to 50.
Check 3: all_cities_present function that verifies city count equals 3.

Materialize assets and verify the checks pass

Your checks are defined, but they have not run yet. Asset checks execute automatically when you materialize the asset they are attached to.

Return to the Dagster UI at http://localhost:3000.
Click Reload definitions to pick up all three checks.

Before you materialize, what do you expect to happen? Do you think all three checks will pass, or will any fail?

Click Materialize all to run both assets and all three checks.
Wait for the run to complete. Both assets should turn green.
Click on the daily_weather_summary asset in the graph to open its details page.
Select the Checks tab.

You should see all three checks listed with green passing indicators. Each check displays its validation metadata: null_temperature_count: 0, out_of_range_count: 0, and city_count: 3.

Why do checks matter in production?

Without checks, your pipeline reports success as long as the code runs without crashing. That means a corrupted API response or a dropped city would produce wrong summaries with no warning.

Asset checks flip that model. They make data quality a first-class signal visible in the UI. If a check fails, you see it immediately instead of discovering the problem weeks later in a downstream report.

Checks not running or showing as failed?

If checks did not appear at all, make sure you clicked Reload definitions before materializing. New files are not picked up automatically.
If a check failed with a database error, verify your PostgreSQL container is still running. Run docker ps in a separate terminal to confirm.
If you see an import error in the dg dev terminal, double check that the import line reads from .assets import daily_weather_summary, DATABASE_URL with a leading dot.

Still stuck? Help me debug why my Dagster asset checks are not passing.

Your pipeline now validates its own output after every run. Green checkmarks confirm the data is correct. Next up, you will schedule the pipeline to run automatically so it does not need you to click buttons.

Schedule the Pipeline

Your pipeline now has automated quality checks that catch bad data before it becomes a problem. There is one gap left: you still have to click Materialize All by hand every time you want fresh data.

A schedule tells Dagster to run the full pipeline automatically at a set time every day. In this step, you will create a schedule that refreshes your weather data every morning at 6 AM without any manual intervention.

In this step, get ready to:

Create a schedule definition that runs the pipeline daily at 6 AM.
Toggle the schedule on in the Dagster UI and verify it with a test run.

Create the schedule definition

A ScheduleDefinition in Dagster pairs a cron expression with a set of assets to materialize. Dagster's auto-discovery system picks up any definition placed in the defs/ folder, so all you need to do is create the file.

Create a new file called schedules.py inside src/weather_pipeline/defs/ and add the following code:

import dagster as dg

# Schedule the full pipeline to run daily at 6 AM
daily_weather_refresh = dg.ScheduleDefinition(
    name="daily_weather_refresh",
    cron_schedule="0 6 * * *",
    target="*",
)

What does this code do?

ScheduleDefinition creates a named schedule that Dagster tracks and displays in its UI.
cron_schedule="0 6 * * *" uses cron syntax to mean "run at 6:00 AM every day." The five fields are: minute, hour, day of month, month, and day of week.
target="*" tells the schedule to materialize all assets in the project, including their checks.

What is a cron expression?

A cron expression is a string of five fields that defines when a task runs. Each field represents minute, hour, day of month, month, and day of week. An asterisk (*) means "every."

So 0 6 * * * reads as: at minute 0 of hour 6, every day of every month, on every day of the week. In plain language: 6:00 AM daily.

Save schedules.py.

Dagster needs to reload its definitions to pick up the new file.

Switch back to the Dagster UI at http://localhost:3000.
Click Deployment in the top navigation bar.
In the Code locations tab, click Reload definitions.
Click Automation in the top navigation bar.
You should see daily_weather_refresh listed as a schedule.

Don't see the schedule in the Automation tab?

Make sure you saved the file at the correct path: src/weather_pipeline/defs/schedules.py.
Confirm that dg dev is still running in your terminal. If it stopped, restart it.
Click Reload definitions again under Deployment to force Dagster to re-read your code. If the code location shows a Failed status, click it to view the error and stack trace.
Help me debug my missing schedule.

✔️ Awesome, I've got everything!

Great. Make sure you have saved schedules.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete schedules.py file:

import dagster as dg

daily_weather_refresh = dg.ScheduleDefinition(
    name="daily_weather_refresh",
    cron_schedule="0 6 * * *",
    target="*",
)

Activate the schedule and trigger a test run

The schedule exists in your code, but it starts in a stopped state by default. You need to toggle it on so Dagster's daemon begins evaluating it.

In the Automation tab, find the daily_weather_refresh schedule.
Click the toggle in the Running column to enable the schedule.

The schedule is now active. You should see the Next tick column showing the next time the schedule will fire (6:00 AM tomorrow).

To verify the schedule works without waiting until tomorrow, trigger a test evaluation.

Click the daily_weather_refresh schedule name to open its detail page.
Click the Preview tick result button near the top right corner of the page.
Select a mock evaluation time from the dropdown and click Continue.

Before you launch the test run, what do you expect to happen? Think about which assets will materialize and in what order.

Click Launch all & commit at the bottom right to launch the run.

Dagster takes you to the Runs page filtered to the launched run. You should see both raw_weather_forecast and daily_weather_summary materialize in dependency order, with all three asset checks passing.

Why use Preview tick result instead of waiting?

Your schedule is set to run at 6 AM, so waiting for it to fire naturally could take hours. The Preview tick result feature lets you simulate a scheduled evaluation at any point in the cadence and launch the resulting run immediately.

Schedule run failed or assets did not materialize?

Make sure the Docker PostgreSQL container is still running. Run docker ps in a terminal to confirm.
Check that target="*" is set in your schedule definition. Without this, the schedule has no assets to materialize.
Click the failed run to view the error log. Common issues include a database connection timeout or a syntax error in schedules.py.
Help me debug my schedule test run.

Secret mission

Build a City Weather Ranking

Your pipeline produces daily summaries, but real pipelines go further. They produce insights. Can you add a ranking asset that identifies the hottest, wettest, and windiest city each day, extending your asset graph to three layers?

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

A Docker container running PostgreSQL 16 on localhost:5432 (uses minimal disk space, no ongoing cost).
A Dagster development server running locally via dg dev.
A Python virtual environment and project folder on your Desktop with Dagster and dependencies.
The uv package manager installed on your system.

Keep everything running

No action needed. Choose this if you are still actively building on the pipeline or want to experiment with new assets and checks.

Your Docker container and Dagster project stay ready for iteration.
The uv package manager remains installed on your system for future Python projects.

Pause - I'll come back to this later

Shut down running processes to free up memory while keeping all your project files intact.

Press Ctrl+C in the terminal running dg dev to stop the Dagster server.
Stop the PostgreSQL container by running:

docker compose stop

To resume later, start the container and Dagster server again by running:

docker compose start

Then activate your virtual environment and run dg dev to pick up where you left off.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your database data, project files, and the tools you installed.

Press Ctrl+C in the terminal running dg dev to stop the Dagster server.
Remove the Docker container and its data volume by running:

docker compose down -v

Delete the project folder from your Desktop by running:

macOS

rm -rf ~/Desktop/weather-pipeline

Windows

rmdir /s /q %USERPROFILE%\Desktop\weather-pipeline

Remove uv from your system by running:

macOS

rm ~/.local/bin/uv ~/.local/bin/uvx

Windows

rm $HOME\.local\bin\uv.exe
rm $HOME\.local\bin\uvx.exe
rm $HOME\.local\bin\uvw.exe

Nice Work!

Nice work! You just built a production-style data pipeline that pulls live weather forecasts from an API, loads them into PostgreSQL, transforms them into daily summaries, validates data quality automatically, and runs on a schedule.

You've learned how to:

Built a complete ELT pipeline in Dagster using the asset-based model: Python functions as assets, declared dependencies between them, and rich metadata tracking with row counts and data previews.
Added automated asset checks that validate pipeline output after every run. Your checks catch null values, out-of-range temperatures, and missing cities before bad data reaches downstream consumers.
Scheduled the pipeline to run daily at 6 AM with a cron schedule and navigated the Dagster UI to explore lineage graphs, run history, asset metadata, and the Automations tab.
Secret Mission: Built a city weather ranking asset that identifies the hottest, wettest, and windiest city each day, extending the pipeline to a three-layer asset graph.

Ready to quiz yourself?