[docs-beta] Add testable code snippets (#23836)

## Summary & Motivation

This moves our beta docsite code examples into its own folder under
`examples`. See instructions for how to run and test this.

## How I Tested These Changes

Local testing, buildkite.

## Changelog [New | Bug | Docs]

NOCHANGELOG

docs/docs-beta/README.md

@@ -32,7 +32,7 @@ brew install vale
 - `./src` contains custom components, styles, themes, and layouts.
 - `./content-templates` contains the templates for the documentation pages.
 - `./docs/` is the source of truth for the documentation.
-- `./docs/code_examples` contains all code examples for the documentation.
+- `/examples/docs_beta_snippets/docs_beta_snippets/` contains all code examples for the documentation.
 
 The docs are broken down into the following sections:
 
@@ -68,6 +68,23 @@ yarn lint:vale
 yarn lint:fix
 ```
 
+To include code snippets, use the following format:
+
+```
+<CodeExample filePath="path/to/file.py" />
+```
+
+The `filePath` is relative to the `./examples/docs_beta_snippets/docs_beta_snippets/` directory.
+
+At minimum, all `.py` files in the `docs_beta_snippets` directory are tested by attempting to load the Python files.
+You can write additional tests for them in the `docs_beta_snippets_test` folder. See the folder for more information.
+
+To type-check the code snippets during development, run the following command from the Dagster root folder.
+This will run `pyright` on all new/changed files relative to the master branch.
+
+```
+make quick_pyright
+```
 ---
 
 ## Build

docs/docs-beta/src/components/CodeExample.tsx

@@ -13,7 +13,7 @@ const CodeExample: React.FC<CodeExampleProps> = ({filePath, language, title}) =>
 
   React.useEffect(() => {
     // Adjust the import path to start from the docs directory
-    import(`!!raw-loader!/docs/code_examples/${filePath}`)
+    import(`!!raw-loader!/../../examples/docs_beta_snippets/docs_beta_snippets/${filePath}`)
       .then((module) => {
         const lines = module.default.split('\n');
         const mainIndex = lines.findIndex((line) => line.trim().startsWith('if __name__ == '));

examples/docs_beta_snippets/README.md

@@ -0,0 +1,15 @@
+# Docs Beta Examples
+
+This module exists only to enable testing docs snippets/examples in CI, and should not be installed
+otherwise.
+
+## Testing
+
+You can test that all code loads into Python correctly with:
+
+```
+pip install -e .
+pytest
+```
+
+You may include additional test files in `docs_beta_snippets_tests`

docs/docs-beta/docs/code_examples/guides/automation/multi-asset-sensor.py → examples/docs_beta_snippets/docs_beta_snippets/guides/automation/multi-asset-sensor.py

@@ -25,7 +25,10 @@ def target_asset():
 )
 def my_multi_asset_sensor(context: MultiAssetSensorEvaluationContext):
     run_requests = []
-    for asset_key, materialization in context.latest_materialization_records_by_key().items():
+    for (
+        asset_key,
+        materialization,
+    ) in context.latest_materialization_records_by_key().items():
         if materialization:
             run_requests.append(RunRequest(asset_selection=[asset_key]))
             context.advance_cursor({asset_key: materialization})

docs/docs-beta/docs/code_examples/guides/automation/simple-schedule-example.py → examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-schedule-example.py

@@ -9,7 +9,9 @@ def customer_data(): ...
 def sales_report(): ...
 
 
-daily_refresh_job = define_asset_job("daily_refresh", selection=["customer_data", "sales_report"])
+daily_refresh_job = define_asset_job(
+    "daily_refresh", selection=["customer_data", "sales_report"]
+)
 
 # highlight-start
 daily_schedule = ScheduleDefinition(

examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-sensor-example.py

@@ -0,0 +1,50 @@
+import random
+from typing import List
+
+from dagster import (
+    AssetExecutionContext,
+    Definitions,
+    RunRequest,
+    SkipReason,
+    asset,
+    define_asset_job,
+    sensor,
+)
+
+
+@asset
+def my_asset(context: AssetExecutionContext):
+    context.log.info("Hello, world!")
+
+
+my_job = define_asset_job("my_job", selection=[my_asset])
+
+
+# highlight-start
+def check_for_new_files() -> List[str]:
+    if random.random() > 0.5:
+        return ["file1", "file2"]
+    return []
+
+
+@sensor(job=my_job, minimum_interval_seconds=5)
+def new_file_sensor():
+    new_files = check_for_new_files()
+    if new_files:
+        for filename in new_files:
+            yield RunRequest(run_key=filename)
+    else:
+        yield SkipReason("No new files found")
+        # highlight-end
+
+
+defs = Definitions(assets=[my_asset], jobs=[my_job], sensors=[new_file_sensor])
+
+
+if __name__ == "__main__":
+    from dagster import materialize
+
+    new_file_sensor()
+    materialize(
+        [my_asset],
+    )

docs/docs-beta/docs/code_examples/guides/data-assets/passing-data-assets/passing-data-io-manager.py → examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-io-manager.py

@@ -1,9 +1,12 @@
 import pandas as pd
-from dagster import Definitions, asset
 from dagster_duckdb_pandas import DuckDBPandasIOManager
 
+from dagster import Definitions, asset
+
 # highlight-start
-duckdb_io_manager = DuckDBPandasIOManager(database="my_database.duckdb", schema="my_schema")
+duckdb_io_manager = DuckDBPandasIOManager(
+    database="my_database.duckdb", schema="my_schema"
+)
 
 
 @asset

examples/docs_beta_snippets/docs_beta_snippets_tests/test_all_files_load.py

@@ -0,0 +1,26 @@
+import importlib.util
+import os
+
+import pytest
+
+from dagster import file_relative_path
+
+snippets_folder = file_relative_path(__file__, "../docs_beta_snippets/")
+
+
+def get_python_files(directory):
+    for root, _, files in os.walk(directory):
+        for file in files:
+            if file.endswith(".py"):
+                yield os.path.join(root, file)
+
+
+@pytest.mark.parametrize("file_path", get_python_files(snippets_folder))
+def test_file_loads(file_path):
+    spec = importlib.util.spec_from_file_location("module", file_path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    try:
+        spec.loader.exec_module(module)
+    except Exception as e:
+        pytest.fail(f"Failed to load {file_path}: {e!s}")

examples/docs_beta_snippets/pyproject.toml

@@ -0,0 +1,35 @@
+[tool.ruff]
+
+# Extend example root configuration.
+extend = "../pyproject.toml"
+
+# Shorter line length for docs snippets for better browser formatting.
+line-length = 88
+
+[tool.ruff.lint]
+
+# Use extend-ignore so that we ignore all the same codes ignored in root.
+extend-ignore = [
+
+  # (Unused import): When the same symbol is imported in multiple blocks, the
+  # last import takes precedence for python. This causes Ruff to think an
+  # import in an earlier block is unused and report F401.
+  "F401",
+
+  # (Redefinition): This happens frequently in docs_snippets when we import the same symbol in multiple
+  # snippets within the same file.
+  "F811",
+
+  # (local variable assigned but never used): This happens a lot in docs snippets for didactic
+  # purposes.
+  "F841",
+
+  # (flake8-type-checking) No need to complicate docs snippets with `if TYPE_CHECKING` blocks.
+  "TCH",
+]
+
+[tool.ruff.lint.isort]
+
+# Ensures ruff classifies imports from `dagster` as first-party. Keeps snippet imports relatively
+# compressed.
+known-first-party = ["dagster"]

examples/docs_beta_snippets/setup.py

@@ -0,0 +1,23 @@
+from setuptools import find_packages, setup
+
+setup(
+    name="docs_beta_snippets",
+    author="Dagster Labs",
+    author_email="[email protected]",
+    license="Apache-2.0",
+    url="https://github.com/dagster-io/dagster/tree/master/examples/docs_beta_snippets",
+    classifiers=[
+        "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+        "License :: OSI Approved :: Apache Software License",
+        "Operating System :: OS Independent",
+    ],
+    packages=find_packages(exclude=["test"]),
+    install_requires=["dagster-cloud"],
+    extras_require={
+        "test": [
+            "pytest",
+        ]
+    },
+)

examples/docs_beta_snippets/tox.ini

@@ -0,0 +1,26 @@
+[tox]
+skipsdist = true
+
+[testenv]
+download = True
+passenv =
+    CI_*
+    COVERALLS_REPO_TOKEN
+    POSTGRES_TEST_DB_HOST
+    BUILDKITE*
+install_command = uv pip install {opts} {packages}
+deps =
+  source: -e ../../python_modules/dagster[test]
+  source: -e ../../python_modules/dagster-pipes
+  pypi: dagster[test]
+  -e ../../python_modules/dagster-graphql
+  -e ../../python_modules/dagster-webserver
+  -e ../../python_modules/libraries/dagster-duckdb-pandas
+  -e ../../python_modules/libraries/dagster-embedded-elt
+  -e .[test]
+allowlist_externals =
+  /bin/bash
+  uv
+commands =
+  source: /bin/bash -c '! pip list --exclude-editable | grep -e dagster'
+  pytest -c ../../pyproject.toml -vv {posargs}

examples/docs_snippets/docs_snippets/docs_beta/guides/automation/asset-sensor-custom-eval.py

@@ -0,0 +1,51 @@
+from dagster import (
+    AssetExecutionContext,
+    AssetKey,
+    AssetMaterialization,
+    Definitions,
+    MaterializeResult,
+    RunRequest,
+    SensorEvaluationContext,
+    SkipReason,
+    asset,
+    asset_sensor,
+    define_asset_job,
+)
+
+
+@asset
+def daily_sales_data(context: AssetExecutionContext):
+    context.log.info("Asset to watch, perhaps some function sets metadata here")
+    yield MaterializeResult(metadata={"specific_property": "value"})
+
+
+@asset
+def weekly_report(context: AssetExecutionContext):
+    context.log.info("Running weekly report")
+
+
+my_job = define_asset_job("my_job", [weekly_report])
+
+
+@asset_sensor(asset_key=AssetKey("daily_sales_data"), job=my_job)
+def daily_sales_data_sensor(context: SensorEvaluationContext, asset_event):
+    # Provide a type hint on the underlying event
+    materialization: AssetMaterialization = (
+        asset_event.dagster_event.event_specific_data.materialization
+    )
+
+    # Example custom logic: Check if the asset metadata has a specific property
+    # highlight-start
+    if "specific_property" in materialization.metadata:
+        context.log.info("Triggering job based on custom evaluation logic")
+        yield RunRequest(run_key=context.cursor)
+    else:
+        yield SkipReason("Asset materialization does not have the required property")
+    # highlight-end
+
+
+defs = Definitions(
+    assets=[daily_sales_data, weekly_report],
+    jobs=[my_job],
+    sensors=[daily_sales_data_sensor],
+)

examples/docs_snippets/docs_snippets/docs_beta/guides/automation/asset-sensor-with-config.py

@@ -0,0 +1,66 @@
+from dagster import (
+    AssetExecutionContext,
+    AssetKey,
+    AssetMaterialization,
+    Config,
+    Definitions,
+    MaterializeResult,
+    RunConfig,
+    RunRequest,
+    SensorEvaluationContext,
+    asset,
+    asset_sensor,
+    define_asset_job,
+)
+
+
+class MyConfig(Config):
+    param1: str
+
+
+@asset
+def daily_sales_data(context: AssetExecutionContext):
+    context.log.info("Asset to watch")
+    # highlight-next-line
+    yield MaterializeResult(metadata={"specific_property": "value"})
+
+
+@asset
+def weekly_report(context: AssetExecutionContext, config: MyConfig):
+    context.log.info(f"Running weekly report with param1: {config.param1}")
+
+
+my_job = define_asset_job(
+    "my_job",
+    [weekly_report],
+    config=RunConfig(ops={"weekly_report": MyConfig(param1="value")}),
+)
+
+
+@asset_sensor(asset_key=AssetKey("daily_sales_data"), job=my_job)
+def daily_sales_data_sensor(context: SensorEvaluationContext, asset_event):
+    materialization: AssetMaterialization = (
+        asset_event.dagster_event.event_specific_data.materialization
+    )
+
+    # Example custom logic: Check if the asset metadata has a specific property
+    # highlight-start
+    if "specific_property" in materialization.metadata:
+        yield RunRequest(
+            run_key=context.cursor,
+            run_config=RunConfig(
+                ops={
+                    "weekly_report": MyConfig(
+                        param1=str(materialization.metadata.get("specific_property"))
+                    )
+                }
+            ),
+        )
+    # highlight-end
+
+
+defs = Definitions(
+    assets=[daily_sales_data, weekly_report],
+    jobs=[my_job],
+    sensors=[daily_sales_data_sensor],
+)