The @resource decorator is used to define a resource - a scoped way to make external resources (like database connections) available to ops and assets during job execution.
Signature
@resource(
config_schema: Optional[ConfigSchema] = None,
description: Optional[str] = None,
required_resource_keys: Optional[AbstractSet[str]] = None,
version: Optional[str] = None,
) -> ResourceDefinition
Parameters
The schema for the config. Configuration data available in init_context.resource_config. If not set, Dagster will accept any config provided.
A human-readable description of the resource.
Keys for the resources required by this resource.
The version of a resource function. Two wrapped resource functions should only have the same version if they produce the same resource definition when provided with the same inputs.
Returns
Type: ResourceDefinition
A resource definition object.
Examples
Basic Resource
from dagster import resource
@resource
def my_resource(init_context):
return "my_resource_value"
Resource with Configuration
from dagster import resource, Field, String
@resource(
config_schema={
"host": Field(String, description="Database host"),
"port": Field(int, default_value=5432),
}
)
def database_resource(init_context):
host = init_context.resource_config["host"]
port = init_context.resource_config["port"]
return DatabaseConnection(host=host, port=port)
Resource with Pythonic Config
from dagster import resource, ConfigurableResource
from pydantic import Field
class DatabaseConfig(ConfigurableResource):
host: str = Field(description="Database host")
port: int = Field(default=5432, description="Database port")
username: str
password: str
def get_connection(self):
return DatabaseConnection(
host=self.host,
port=self.port,
username=self.username,
password=self.password,
)
Resource with Cleanup (Context Manager)
from dagster import resource
from contextlib import contextmanager
@resource
def database_resource(init_context):
# Setup: create connection
connection = DatabaseConnection()
try:
yield connection
finally:
# Teardown: close connection
connection.close()
Resource with Dependencies
from dagster import resource
@resource
def credentials_resource(init_context):
return {"api_key": "secret_key"}
@resource(required_resource_keys={"credentials"})
def api_client_resource(init_context):
credentials = init_context.resources.credentials
return APIClient(api_key=credentials["api_key"])
Using Resources in Ops
from dagster import op, job, resource, OpExecutionContext
@resource
def database_resource(init_context):
return DatabaseConnection()
@op(required_resource_keys={"database"})
def query_database(context: OpExecutionContext):
db = context.resources.database
return db.query("SELECT * FROM users")
@job(resource_defs={"database": database_resource})
def database_job():
query_database()
Using Resources in Assets
from dagster import asset, Definitions, ConfigurableResource
class MyDatabaseResource(ConfigurableResource):
connection_string: str
def query(self, sql: str):
# Execute query
return results
@asset
def my_asset(database: MyDatabaseResource):
# Resource is automatically provided
return database.query("SELECT * FROM table")
defs = Definitions(
assets=[my_asset],
resources={
"database": MyDatabaseResource(
connection_string="postgresql://localhost/mydb"
)
},
)
Hardcoded Resource (for Testing)
from dagster import ResourceDefinition
# For production
production_db = database_resource.configured({
"host": "prod.database.com",
"port": 5432,
})
# For testing - no initialization needed
test_db = ResourceDefinition.hardcoded_resource(
MockDatabase(),
description="Mock database for testing"
)
Mock Resource
from dagster import ResourceDefinition
# Creates a resource that returns a MagicMock
mock_api = ResourceDefinition.mock_resource(
description="Mock API client for testing"
)
String Resource Helper
from dagster import ResourceDefinition
# Simple resource that just stores a string value
api_key_resource = ResourceDefinition.string_resource(
description="API key for external service"
)
# Used with config
api_key_resource.configured("my-secret-api-key")
