Implementation:Dagster io Dagster ConfigurableResource Base

Field	Value
Implementation Name	ConfigurableResource Base
Source	`python_modules/dagster/dagster/_config/pythonic_config/resource.py:L599`
Repository	dagster-io/dagster
Domains	Data_Engineering, API_Integration

Overview

Concrete base class for building custom Dagster resources with Pydantic-based configuration provided by the Dagster core library.

Description

ConfigurableResource is the base class in Dagster for creating custom resources that utilize structured config. It is a subclass of both ResourceDefinition and Pydantic's config system, allowing resources to declare typed configuration fields that are validated at definition time and visible in the Dagster UI. Subclasses define Pydantic fields for configuration parameters and implement methods for obtaining authenticated clients or performing service-specific operations.

Usage

Subclass ConfigurableResource to create a resource for any external service. Define Pydantic fields for configuration, use EnvVar for secrets, and implement methods that return configured client instances or perform operations. Register the resource in Definitions with a key name, and inject it into assets by using that key as a parameter name.

Code Reference

Source Location

python_modules/dagster/dagster/_config/pythonic_config/resource.py:L599

Base Class

class ConfigurableResource(ConfigurableResourceFactory[TResValue]):
    """Base class for Dagster resources that utilize structured config.

    This class is a subclass of both ResourceDefinition and Config.
    """
    pass

Common Patterns

PineconeResource Example

import dagster as dg
from pydantic import Field

class PineconeResource(dg.ConfigurableResource):
    pinecone_api_key: str = Field(description="Pinecone API key")
    index_name: str = Field(default="my-index", description="Index name")

    def create_index(self, dimension: int = 1536):
        from pinecone import Pinecone
        pc = Pinecone(api_key=self.pinecone_api_key)
        pc.create_index(name=self.index_name, dimension=dimension, metric="cosine")

    def get_index(self, namespace: str):
        from pinecone import Pinecone
        pc = Pinecone(api_key=self.pinecone_api_key)
        return pc.Index(self.index_name).namespace(namespace)

DSPyResource Example

import dagster as dg
from pydantic import Field

class DSPyResource(dg.ConfigurableResource):
    model_name: str = Field(default="gemini-2.0-flash-exp")
    api_key: str = Field(description="Gemini API key")
    model_dir: str = Field(default="models/", description="Directory for saving models")

    def configure_dspy(self):
        import dspy
        lm = dspy.LM(model=f"gemini/{self.model_name}", api_key=self.api_key)
        dspy.configure(lm=lm)

    def save_model(self, model, path: str):
        model.save(path)

    def load_model(self, model_class, path: str):
        model = model_class()
        model.load(path)
        return model

ATProtoResource Example

import dagster as dg
from atproto import Client

class ATProtoResource(dg.ConfigurableResource):
    login: str
    password: str
    session_cache_path: str = "atproto-session.txt"

    def get_client(self) -> Client:
        client = Client()
        self._login(client)
        return client

Import

from dagster import ConfigurableResource or import dagster as dg; class MyResource(dg.ConfigurableResource)

I/O Contract

Direction	Name	Type	Description
Input	Pydantic fields	`str`, `int`, `float`, etc.	Typed configuration fields declared as class attributes with optional `Field` descriptors
Input	secrets	`EnvVar`	Environment variable references for sensitive values (API keys, passwords)
Output	resource instance	custom class	Configured resource instance injected into assets via parameter name matching
Output	client objects	varies	Authenticated client instances returned by resource methods (e.g., `get_client()`, `get_index()`)

Usage Examples

Definitions Registration

import dagster as dg

defs = dg.Definitions(
    resources={
        "pinecone": PineconeResource(
            pinecone_api_key=dg.EnvVar("PINECONE_API_KEY"),
            index_name="dagster-knowledge",
        ),
        "dspy": DSPyResource(
            api_key=dg.EnvVar("GEMINI_API_KEY"),
            model_name="gemini-2.0-flash-exp",
            model_dir="models/",
        ),
    },
    assets=[...],
)

Asset Using Custom Resource

import dagster as dg

@dg.asset(kinds={"pinecone"})
def upsert_embeddings(pinecone: PineconeResource) -> dg.MaterializeResult:
    index = pinecone.get_index(namespace="docs")
    # ... upsert vectors ...
    return dg.MaterializeResult(metadata={"vectors_upserted": 1000})

@dg.asset(kinds={"dspy"})
def train_model(dspy_resource: DSPyResource) -> dg.MaterializeResult:
    dspy_resource.configure_dspy()
    # ... train model ...
    dspy_resource.save_model(model, "models/my_model.json")
    return dg.MaterializeResult(metadata={"model_path": "models/my_model.json"})

Related Pages

Principle:Dagster_io_Dagster_Custom_Resource_Pattern

Requires Environment

Environment:Dagster_io_Dagster_Python_3_10_Runtime

Uses Heuristic

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment