Dagster PostgreSQL Integration

0.28.22 · active · verified Thu Apr 09

dagster-postgres is an integration library for Dagster that provides resources and storage implementations for interacting with PostgreSQL databases. It allows Dagster assets to connect to Postgres, and can also use Postgres for Dagster's run storage and event log storage. It is currently at version 0.28.22 and follows the rapid release cycle of the main Dagster project.

Warnings

gotcha The `psycopg2-binary` package was removed as a transitive dependency in `dagster-postgres` version 0.28.18 (Dagster core 1.12.18). If your project relied on this transitive dependency, you must now explicitly install `psycopg2-binary` (or `psycopg2`) in your environment.
Fix: Explicitly add `psycopg2-binary` or `psycopg2` to your project's dependencies: `pip install psycopg2-binary`
gotcha Dagster library versions (e.g., `dagster-postgres`) are tightly coupled with the core `dagster` library version. Mismatched versions can lead to unexpected behavior or runtime errors. Always ensure all `dagster-*` packages in your environment match the major and minor version of your core `dagster` installation.
Fix: Always install `dagster-postgres` with the same major.minor version as your core `dagster` package (e.g., if `dagster==1.12.x`, install `dagster-postgres==0.28.x`). Use `pip install dagster==X.Y.Z dagster-postgres==0.X.Y`.
gotcha When configuring `postgres_resource`, sensitive credentials (user, password) should always be managed securely, ideally using environment variables or a secret management system, rather than hardcoding them directly in code or `dagster.yaml`.
Fix: Use `EnvVar()` for programmatic configuration or `${MY_SECRET_ENV_VAR}` syntax in `dagster.yaml` to pull credentials from environment variables.

Install

pip install dagster-postgres Install library

Imports

postgres_resource

from dagster_postgres import postgres_resource

PostgresRunStorage
```
from dagster_postgres.run_storage import PostgresRunStorage
```
Used for configuring Dagster's run storage in `dagster.yaml`.
PostgresEventLogStorage
```
from dagster_postgres.event_log import PostgresEventLogStorage
```
Used for configuring Dagster's event log storage in `dagster.yaml`.

Quickstart

This example defines a simple Dagster asset that uses `postgres_resource` to connect to a PostgreSQL database. It demonstrates how to configure the resource using environment variables and then obtain a connection to execute a query. Ensure you have the necessary PostgreSQL environment variables set for the connection to succeed.

import os
from dagster import Definitions, asset, EnvVar
from dagster_postgres import postgres_resource

@asset(compute_kind="sql")
def my_postgres_asset(postgres: postgres_resource):
    """A simple asset that connects to PostgreSQL and runs a dummy query."""
    with postgres.get_connection() as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT 1;")
            result = cursor.fetchone()
            print(f"Postgres query result: {result}")
            # In a typical asset, you'd load/transform/store data here.

defs = Definitions(
    assets=[my_postgres_asset],
    resources={
        "postgres": postgres_resource.configured({
            "database": EnvVar("POSTGRES_DB"),
            "host": EnvVar("POSTGRES_HOST"),
            "port": EnvVar("POSTGRES_PORT"),
            "user": EnvVar("POSTGRES_USER"),
            "password": EnvVar("POSTGRES_PASSWORD"),
        })
    }
)

# To run this example, set environment variables (e.g., in your shell):
# export POSTGRES_DB=your_db_name
# export POSTGRES_HOST=localhost
# export POSTGRES_PORT=5432
# export POSTGRES_USER=your_user
# export POSTGRES_PASSWORD=your_password
# Then execute using `dagster dev -f your_file.py` and materialize `my_postgres_asset` in the UI.

view raw JSON →