Using · Dagster Expert · dagster-io/skills · Skills Docs

Using · Dagster Expert · dagster-io/skills · Skills Docs

Strategy	Storage	Refresh Mechanism	Best For
`LOCAL_FILESYSTEM`	`.local_defs_state/` in project dir	`dg utils refresh-defs-state` in CI before building artifact	Most deployments
`VERSIONED_STATE_STORAGE`	Cloud storage (S3 on Dagster+, or S3/GCS/Azure)	`dg utils refresh-defs-state` or independent of deploys	Dagster+ (automatic); decoupling state from builds
`LEGACY_CODE_SERVER_SNAPSHOTS`	In-memory	Auto-refresh on every code server load	Not recommended; backwards-compat only

my_project/
├── my_project/
│   ├── defs/
│   └── ...
├── .local_defs_state/
│   ├── DbtProjectComponent[my_dbt_project]/
│   └── FivetranAccountComponent/
├── pyproject.toml
└── .gitignore          # includes .local_defs_state/

# dagster.yaml (OSS / self-hosted only — not needed on Dagster+)
defs_state_storage:
  module: dagster._core.storage.defs_state.blob_storage_state_storage
  class: UPathDefsStateStorage
  config:
    base_path: s3://my-bucket/dagster/defs-state # or gs:// or az://

Field	Type	Default	Description
`management_type`	`LOCAL_FILESYSTEM` \| `VERSIONED_STATE_STORAGE` \| `LEGACY_CODE_SERVER_SNAPSHOTS`	Component-specific	Storage strategy for state
`key`	string	Auto-generated from class name	Unique identifier for this component’s state
`refresh_if_dev`	boolean	`true`	Auto-refresh state during `dagster dev`. Set `false` to reduce API calls during local dev

import asyncio
import dagster as dg


class MyApiComponent(dg.StateBackedComponent, dg.Model, dg.Resolvable):
    api_url: str

    # ... defs_state_config, write_state_to_path, build_defs_from_state ...

    def build_defs_from_state(
        self, context: dg.ComponentLoadContext, state_path: Path | None
    ) -> dg.Definitions:
        assets = ...  # build assets from state
        op_name = f"refresh_state_{hash(self.defs_state_config.key) % 10**8}"

        @dg.op(name=op_name)
        def refresh_my_api_state():
            asyncio.run(self.refresh_state(context.project_root))

        @dg.job(name="refresh_my_api_component_state")
        def refresh_job():
            refresh_my_api_state()

        return dg.Definitions(assets=assets, jobs=[refresh_job])

mutation RefreshDefsStateMutation($locationName: String!) {
  refreshDefsState(locationName: $locationName) {
    ... on WorkspaceEntry {
      locationName
    }
    ... on RefreshDefsStateError {
      message
    }
  }
}