From 9dcf00e241591f691d8c8853c37eceb25b419037 Mon Sep 17 00:00:00 2001
From: Pedram Navid <1045990+PedramNavid@users.noreply.github.com>
Date: Sat, 24 Aug 2024 19:31:27 -0700
Subject: [PATCH] [docs-beta] Update to dagster as dg (#23864)

## Summary & Motivation

Update the beta docs code snippets to use the new `import dagster as dg` code pattern. This should be the new approach to all documentation that references dagster code, as consistency is important.

Here are a few reasons why I like this approach:

0) There is prior art in many libraries of this approach, notably `pandas`, `sklearn`, `numpy`, `matplotlib`, `pytest` and many other packages that export many objects.

1) Fewer import lines in code samples/snippets. Compare the first image which is 12 lines shorter. This makes code easier for users to understand parse.

![image.png](https://graphite-user-uploaded-assets-prod.s3.amazonaws.com/80phFRWNTJBzbgP9Xj13/1c9cc21d-3b04-4eab-b174-77d23cf74d63.png)

2) Another side benefit is that type-ahead experience is improved.

In VS Code, for example, typing `dg.sensor` and pressing Ctrl+Space provides you with all the exported Sensor-like objects

![image.png](https://graphite-user-uploaded-assets-prod.s3.amazonaws.com/80phFRWNTJBzbgP9Xj13/3c0776cb-8006-488e-849a-36ae64c193a1.png)

Compare to typing `sensor` <ctrl-space> instead with `import dagster`

![image.png](https://graphite-user-uploaded-assets-prod.s3.amazonaws.com/80phFRWNTJBzbgP9Xj13/8de32f3e-7ac3-4902-84a5-ebba88ebea24.png)

3) Finally, we can always revert this change in the future by searching the examples for `dg.` if we decide this approach is fundamentally broken.

## How I Tested These Changes

bk, pytest

## Changelog [New | Bug | Docs]

NOCHANGELOG
---
 examples/docs_beta_snippets/README.md         | 15 +++++++
 .../automation/asset-sensor-custom-eval.py    | 38 ++++++----------
 .../automation/asset-sensor-with-config.py    | 43 +++++++------------
 .../guides/automation/multi-asset-sensor.py   | 23 ++++------
 .../automation/schedule-with-partition.py     | 15 +++----
 .../automation/simple-asset-sensor-example.py | 26 ++++-------
 .../automation/simple-schedule-example.py     | 12 +++---
 .../automation/simple-sensor-example.py       | 32 +++++---------
 .../passing-data-assets/passing-data-avoid.py |  8 ++--
 .../passing-data-explicit.py                  | 14 +++---
 .../passing-data-io-manager.py                | 14 +++---
 .../passing-data-rewrite-assets.py            |  4 +-
 .../metadata/custom-local-references.py       | 16 +++----
 .../metadata/definition-metadata.py           | 14 +++---
 .../data-modeling/metadata/oss-references.py  | 20 +++------
 .../guides/data-modeling/metadata/owners.py   |  6 +--
 .../data-modeling/metadata/plus-references.py | 10 ++---
 .../metadata/python-local-references.py       | 10 ++---
 .../metadata/runtime-metadata.py              |  8 ++--
 .../metadata/table-column-lineage-metadata.py | 26 +++++------
 .../metadata/table-schema-metadata.py         | 18 ++++----
 .../guides/data-modeling/metadata/tags.py     |  6 +--
 22 files changed, 158 insertions(+), 220 deletions(-)

diff --git a/examples/docs_beta_snippets/README.md b/examples/docs_beta_snippets/README.md
index 5484ad3550409..14d54d333cac6 100755
--- a/examples/docs_beta_snippets/README.md
+++ b/examples/docs_beta_snippets/README.md
@@ -3,6 +3,21 @@
 This module exists only to enable testing docs snippets/examples in CI, and should not be installed
 otherwise.
 
+## Code Style
+
+Please use the following code style to keep imports short:
+
+```python
+import dagster as dg
+
+def my_cool_asset(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
+    return dg.MaterializeResult(
+        metadata={
+            "foo": "bar",
+        }
+    )
+```
+
 ## Testing
 
 You can test that all code loads into Python correctly with:
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/asset-sensor-custom-eval.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/asset-sensor-custom-eval.py
index 8a0f006206861..dd417c9fe4dc3 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/asset-sensor-custom-eval.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/asset-sensor-custom-eval.py
@@ -1,36 +1,24 @@
-from dagster import (
-    AssetExecutionContext,
-    AssetKey,
-    AssetMaterialization,
-    Definitions,
-    MaterializeResult,
-    RunRequest,
-    SensorEvaluationContext,
-    SkipReason,
-    asset,
-    asset_sensor,
-    define_asset_job,
-)
+import dagster as dg
 
 
-@asset
-def daily_sales_data(context: AssetExecutionContext):
+@dg.asset
+def daily_sales_data(context: dg.AssetExecutionContext):
     context.log.info("Asset to watch, perhaps some function sets metadata here")
-    yield MaterializeResult(metadata={"specific_property": "value"})
+    yield dg.MaterializeResult(metadata={"specific_property": "value"})
 
 
-@asset
-def weekly_report(context: AssetExecutionContext):
+@dg.asset
+def weekly_report(context: dg.AssetExecutionContext):
     context.log.info("Running weekly report")
 
 
-my_job = define_asset_job("my_job", [weekly_report])
+my_job = dg.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):
+@dg.asset_sensor(asset_key=dg.AssetKey("daily_sales_data"), job=my_job)
+def daily_sales_data_sensor(context: dg.SensorEvaluationContext, asset_event):
     # Provide a type hint on the underlying event
-    materialization: AssetMaterialization = (
+    materialization: dg.AssetMaterialization = (
         asset_event.dagster_event.event_specific_data.materialization
     )
 
@@ -38,13 +26,13 @@ def daily_sales_data_sensor(context: SensorEvaluationContext, asset_event):
     # highlight-start
     if "specific_property" in materialization.metadata:
         context.log.info("Triggering job based on custom evaluation logic")
-        yield RunRequest(run_key=context.cursor)
+        yield dg.RunRequest(run_key=context.cursor)
     else:
-        yield SkipReason("Asset materialization does not have the required property")
+        yield dg.SkipReason("Asset materialization does not have the required property")
     # highlight-end
 
 
-defs = Definitions(
+defs = dg.Definitions(
     assets=[daily_sales_data, weekly_report],
     jobs=[my_job],
     sensors=[daily_sales_data_sensor],
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/asset-sensor-with-config.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/asset-sensor-with-config.py
index 3c729f898acc5..a9053ef131ebc 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/asset-sensor-with-config.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/asset-sensor-with-config.py
@@ -1,54 +1,41 @@
-from dagster import (
-    AssetExecutionContext,
-    AssetKey,
-    AssetMaterialization,
-    Config,
-    Definitions,
-    MaterializeResult,
-    RunConfig,
-    RunRequest,
-    SensorEvaluationContext,
-    asset,
-    asset_sensor,
-    define_asset_job,
-)
+import dagster as dg
 
 
-class MyConfig(Config):
+class MyConfig(dg.Config):
     param1: str
 
 
-@asset
-def daily_sales_data(context: AssetExecutionContext):
+@dg.asset
+def daily_sales_data(context: dg.AssetExecutionContext):
     context.log.info("Asset to watch")
     # highlight-next-line
-    yield MaterializeResult(metadata={"specific_property": "value"})
+    yield dg.MaterializeResult(metadata={"specific_property": "value"})
 
 
-@asset
-def weekly_report(context: AssetExecutionContext, config: MyConfig):
+@dg.asset
+def weekly_report(context: dg.AssetExecutionContext, config: MyConfig):
     context.log.info(f"Running weekly report with param1: {config.param1}")
 
 
-my_job = define_asset_job(
+my_job = dg.define_asset_job(
     "my_job",
     [weekly_report],
-    config=RunConfig(ops={"weekly_report": MyConfig(param1="value")}),
+    config=dg.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 = (
+@dg.asset_sensor(asset_key=dg.AssetKey("daily_sales_data"), job=my_job)
+def daily_sales_data_sensor(context: dg.SensorEvaluationContext, asset_event):
+    materialization: dg.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(
+        yield dg.RunRequest(
             run_key=context.cursor,
-            run_config=RunConfig(
+            run_config=dg.RunConfig(
                 ops={
                     "weekly_report": MyConfig(
                         param1=str(materialization.metadata.get("specific_property"))
@@ -59,7 +46,7 @@ def daily_sales_data_sensor(context: SensorEvaluationContext, asset_event):
     # highlight-end
 
 
-defs = Definitions(
+defs = dg.Definitions(
     assets=[daily_sales_data, weekly_report],
     jobs=[my_job],
     sensors=[daily_sales_data_sensor],
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/multi-asset-sensor.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/multi-asset-sensor.py
index d3f5adffc25b1..59cc2df3c8e33 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/multi-asset-sensor.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/multi-asset-sensor.py
@@ -1,35 +1,28 @@
-from dagster import (
-    AssetKey,
-    MultiAssetSensorEvaluationContext,
-    RunRequest,
-    asset,
-    define_asset_job,
-    multi_asset_sensor,
-)
+import dagster as dg
 
 
-@asset
+@dg.asset
 def target_asset():
     pass
 
 
-downstream_job = define_asset_job("downstream_job", [target_asset])
+downstream_job = dg.define_asset_job("downstream_job", [target_asset])
 
 
-@multi_asset_sensor(
+@dg.multi_asset_sensor(
     monitored_assets=[
-        AssetKey("upstream_asset_1"),
-        AssetKey("upstream_asset_2"),
+        dg.AssetKey("upstream_asset_1"),
+        dg.AssetKey("upstream_asset_2"),
     ],
     job=downstream_job,
 )
-def my_multi_asset_sensor(context: MultiAssetSensorEvaluationContext):
+def my_multi_asset_sensor(context: dg.MultiAssetSensorEvaluationContext):
     run_requests = []
     for (
         asset_key,
         materialization,
     ) in context.latest_materialization_records_by_key().items():
         if materialization:
-            run_requests.append(RunRequest(asset_selection=[asset_key]))
+            run_requests.append(dg.RunRequest(asset_selection=[asset_key]))
             context.advance_cursor({asset_key: materialization})
     return run_requests
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/schedule-with-partition.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/schedule-with-partition.py
index 2ec9852fb5151..b8f8d8ce6b4f3 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/schedule-with-partition.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/schedule-with-partition.py
@@ -1,22 +1,17 @@
-from dagster import (
-    DailyPartitionsDefinition,
-    asset,
-    build_schedule_from_partitioned_job,
-    define_asset_job,
-)
+import dagster as dg
 
-daily_partition = DailyPartitionsDefinition(start_date="2024-05-20")
+daily_partition = dg.DailyPartitionsDefinition(start_date="2024-05-20")
 
 
-@asset(partitions_def=daily_partition)
+@dg.asset(partitions_def=daily_partition)
 def daily_asset(): ...
 
 
-partitioned_asset_job = define_asset_job("partitioned_job", selection=[daily_asset])
+partitioned_asset_job = dg.define_asset_job("partitioned_job", selection=[daily_asset])
 
 # highlight-start
 # This partition will run daily
-asset_partitioned_schedule = build_schedule_from_partitioned_job(
+asset_partitioned_schedule = dg.build_schedule_from_partitioned_job(
     partitioned_asset_job,
 )
 # highlight-end
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-asset-sensor-example.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-asset-sensor-example.py
index ad7c30f647118..b7e272d21cd43 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-asset-sensor-example.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-asset-sensor-example.py
@@ -1,35 +1,27 @@
-from dagster import (
-    AssetExecutionContext,
-    AssetKey,
-    Definitions,
-    RunRequest,
-    asset,
-    asset_sensor,
-    define_asset_job,
-)
+import dagster as dg
 
 
-@asset
-def daily_sales_data(context: AssetExecutionContext):
+@dg.asset
+def daily_sales_data(context: dg.AssetExecutionContext):
     context.log.info("Asset to watch")
 
 
-@asset
-def weekly_report(context: AssetExecutionContext):
+@dg.asset
+def weekly_report(context: dg.AssetExecutionContext):
     context.log.info("Asset to trigger")
 
 
-my_job = define_asset_job("my_job", [weekly_report])
+my_job = dg.define_asset_job("my_job", [weekly_report])
 
 
 # highlight-start
-@asset_sensor(asset_key=AssetKey("daily_sales_data"), job_name="my_job")
+@dg.asset_sensor(asset_key=dg.AssetKey("daily_sales_data"), job_name="my_job")
 def daily_sales_data_sensor():
-    return RunRequest()
+    return dg.RunRequest()
     # highlight-end
 
 
-defs = Definitions(
+defs = dg.Definitions(
     assets=[daily_sales_data, weekly_report],
     jobs=[my_job],
     sensors=[daily_sales_data_sensor],
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-schedule-example.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-schedule-example.py
index 065fc817dd7fa..a57f8280b15ce 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-schedule-example.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-schedule-example.py
@@ -1,26 +1,26 @@
-from dagster import Definitions, ScheduleDefinition, asset, define_asset_job
+import dagster as dg
 
 
-@asset
+@dg.asset
 def customer_data(): ...
 
 
-@asset
+@dg.asset
 def sales_report(): ...
 
 
-daily_refresh_job = define_asset_job(
+daily_refresh_job = dg.define_asset_job(
     "daily_refresh", selection=["customer_data", "sales_report"]
 )
 
 # highlight-start
-daily_schedule = ScheduleDefinition(
+daily_schedule = dg.ScheduleDefinition(
     job=daily_refresh_job,
     cron_schedule="0 0 * * *",  # Runs at midnight daily
 )
 # highlight-end
 
-defs = Definitions(
+defs = dg.Definitions(
     assets=[customer_data, sales_report],
     jobs=[daily_refresh_job],
     schedules=[daily_schedule],
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-sensor-example.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-sensor-example.py
index 2a132af4e962e..64fbf6ee17702 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-sensor-example.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/automation/simple-sensor-example.py
@@ -1,23 +1,15 @@
 import random
 from typing import List
 
-from dagster import (
-    AssetExecutionContext,
-    Definitions,
-    RunRequest,
-    SkipReason,
-    asset,
-    define_asset_job,
-    sensor,
-)
-
-
-@asset
-def my_asset(context: AssetExecutionContext):
+import dagster as dg
+
+
+@dg.asset
+def my_asset(context: dg.AssetExecutionContext):
     context.log.info("Hello, world!")
 
 
-my_job = define_asset_job("my_job", selection=[my_asset])
+my_job = dg.define_asset_job("my_job", selection=[my_asset])
 
 
 # highlight-start
@@ -27,24 +19,22 @@ def check_for_new_files() -> List[str]:
     return []
 
 
-@sensor(job=my_job, minimum_interval_seconds=5)
+@dg.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)
+            yield dg.RunRequest(run_key=filename)
     else:
-        yield SkipReason("No new files found")
+        yield dg.SkipReason("No new files found")
         # highlight-end
 
 
-defs = Definitions(assets=[my_asset], jobs=[my_job], sensors=[new_file_sensor])
+defs = dg.Definitions(assets=[my_asset], jobs=[my_job], sensors=[new_file_sensor])
 
 
 if __name__ == "__main__":
-    from dagster import materialize
-
     new_file_sensor()
-    materialize(
+    dg.materialize(
         [my_asset],
     )
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-avoid.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-avoid.py
index ab5a24503e309..20d019f0e5365 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-avoid.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-avoid.py
@@ -1,20 +1,20 @@
-from dagster import asset
+import dagster as dg
 
 
 # Warning! This is not the right way to create assets
-@asset
+@dg.asset
 def download_files():
     # Download files from S3, the web, etc.
     ...
 
 
-@asset
+@dg.asset
 def unzip_files():
     # Unzip files to local disk or persistent storage
     ...
 
 
-@asset
+@dg.asset
 def load_data():
     # Read data previously written and store in a data warehouse
     ...
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-explicit.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-explicit.py
index 697ba076212c0..af5100007c34e 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-explicit.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-explicit.py
@@ -1,30 +1,28 @@
 import sqlite3
 import tempfile
 
-from dagster import AssetExecutionContext, Definitions, asset
+import dagster as dg
 
 database_file = tempfile.NamedTemporaryFile()
 
 
 # highlight-start
-@asset
+@dg.asset
 def asset1():
     with sqlite3.connect("database.sqlite") as conn:
         conn.execute("CREATE OR REPLACE TABLE test (i INTEGER)")
         conn.execute("INSERT INTO test VALUES (42)")
 
 
-@asset(deps=[asset1])
-def asset2(context: AssetExecutionContext):
+@dg.asset(deps=[asset1])
+def asset2(context: dg.AssetExecutionContext):
     with sqlite3.connect("database.sqlite") as conn:
         result = conn.execute("SELECT * FROM test").fetchall()
         context.log.info(result)
         # highlight-end
 
 
-defs = Definitions(assets=[asset1, asset2])
+defs = dg.Definitions(assets=[asset1, asset2])
 
 if __name__ == "__main__":
-    from dagster import materialize
-
-    materialize(assets=[asset1, asset2])
+    dg.materialize(assets=[asset1, asset2])
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-io-manager.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-io-manager.py
index de96b17aa873c..ab4b2587d7025 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-io-manager.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-io-manager.py
@@ -1,7 +1,7 @@
 import pandas as pd
 from dagster_duckdb_pandas import DuckDBPandasIOManager
 
-from dagster import Definitions, asset
+import dagster as dg
 
 # highlight-start
 duckdb_io_manager = DuckDBPandasIOManager(
@@ -9,28 +9,26 @@
 )
 
 
-@asset
+@dg.asset
 def people():
     return pd.DataFrame({"id": [1, 2, 3], "name": ["Alice", "Bob", "Charlie"]})
 
 
-@asset
+@dg.asset
 def birds():
     return pd.DataFrame({"id": [1, 2, 3], "name": ["Bluebird", "Robin", "Eagle"]})
 
 
-@asset
+@dg.asset
 def combined_data(people, birds):
     return pd.concat([people, birds])
     # highlight-end
 
 
-defs = Definitions(
+defs = dg.Definitions(
     assets=[people, birds, combined_data],
     resources={"io_manager": duckdb_io_manager},
 )
 
 if __name__ == "__main__":
-    from dagster import materialize
-
-    materialize(assets=[people, birds, combined_data])
+    dg.materialize(assets=[people, birds, combined_data])
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-rewrite-assets.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-rewrite-assets.py
index 753b26a4b4e21..b66fa06fdda1e 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-rewrite-assets.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-assets/passing-data-assets/passing-data-rewrite-assets.py
@@ -1,6 +1,6 @@
 from typing import List
 
-from dagster import asset
+import dagster as dg
 
 
 def download_files() -> str:
@@ -18,7 +18,7 @@ def load_data(files: List[str]):
     ...
 
 
-@asset
+@dg.asset
 def my_dataset():
     zipped_files = download_files()
     files = unzip_files(zipped_files)
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/custom-local-references.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/custom-local-references.py
index 3efe88f0728e0..948dc509da9b0 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/custom-local-references.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/custom-local-references.py
@@ -1,17 +1,11 @@
-from dagster import (
-    CodeReferencesMetadataValue,
-    Definitions,
-    LocalFileCodeReference,
-    asset,
-    with_source_code_references,
-)
+import dagster as dg
 
 
-@asset(
+@dg.asset(
     metadata={
-        "dagster/code_references": CodeReferencesMetadataValue(
+        "dagster/code_references": dg.CodeReferencesMetadataValue(
             code_references=[
-                LocalFileCodeReference(
+                dg.LocalFileCodeReference(
                     file_path="/path/to/source.yaml",
                     # Label and line number are optional
                     line_number=1,
@@ -24,4 +18,4 @@
 def my_asset_modeled_in_yaml(): ...
 
 
-defs = Definitions(assets=with_source_code_references([my_asset_modeled_in_yaml]))
+defs = dg.Definitions(assets=dg.with_source_code_references([my_asset_modeled_in_yaml]))
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/definition-metadata.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/definition-metadata.py
index eee9e69d9684e..23761413978e4 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/definition-metadata.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/definition-metadata.py
@@ -1,21 +1,21 @@
-from dagster import AssetSpec, MetadataValue, asset
+import dagster as dg
 
 
 # You can attach metadata via the `metadata` argument on the `@asset` decorator.
-@asset(
+@dg.asset(
     metadata={
-        "link_to_docs": MetadataValue.url("https://..."),
-        "snippet": MetadataValue.md("# Embedded markdown\n..."),
+        "link_to_docs": dg.MetadataValue.url("https://..."),
+        "snippet": dg.MetadataValue.md("# Embedded markdown\n..."),
     }
 )
 def my_asset(): ...
 
 
 # You can also use `metadata` with `AssetSpec`
-my_external_asset = AssetSpec(
+my_external_asset = dg.AssetSpec(
     "my_external_asset",
     metadata={
-        "link_to_docs": MetadataValue.url("https://..."),
-        "snippet": MetadataValue.md("# Embedded markdown\n..."),
+        "link_to_docs": dg.MetadataValue.url("https://..."),
+        "snippet": dg.MetadataValue.md("# Embedded markdown\n..."),
     },
 )
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/oss-references.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/oss-references.py
index d9ba57917da38..817bff99706e2 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/oss-references.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/oss-references.py
@@ -1,31 +1,25 @@
 import os
 from pathlib import Path
 
-from dagster import (
-    AnchorBasedFilePathMapping,
-    Definitions,
-    asset,
-    link_code_references_to_git,
-    with_source_code_references,
-)
+import dagster as dg
 
 
-@asset
+@dg.asset
 def my_asset(): ...
 
 
-@asset
+@dg.asset
 def another_asset(): ...
 
 
-assets = with_source_code_references([my_asset, another_asset])
+assets = dg.with_source_code_references([my_asset, another_asset])
 
-defs = Definitions(
-    assets=link_code_references_to_git(
+defs = dg.Definitions(
+    assets=dg.link_code_references_to_git(
         assets_defs=assets,
         git_url="https://github.com/dagster-io/dagster",
         git_branch="main",
-        file_path_mapping=AnchorBasedFilePathMapping(
+        file_path_mapping=dg.AnchorBasedFilePathMapping(
             local_file_anchor=Path(__file__),
             file_anchor_path_in_repository="src/repo.py",
         ),
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/owners.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/owners.py
index 0a7d5f3a4efb1..ef6ac4c522a9e 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/owners.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/owners.py
@@ -1,12 +1,12 @@
-from dagster import AssetSpec, asset
+import dagster as dg
 
 
 # You can attach owners via the `owners` argument on the `@asset` decorator.
-@asset(owners=["richard.hendricks@hooli.com", "team:data-eng"])
+@dg.asset(owners=["richard.hendricks@hooli.com", "team:data-eng"])
 def my_asset(): ...
 
 
 # You can also use `owners` with `AssetSpec`
-my_external_asset = AssetSpec(
+my_external_asset = dg.AssetSpec(
     "my_external_asset", owners=["bighead@hooli.com", "team:roof", "team:corpdev"]
 )
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/plus-references.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/plus-references.py
index 309f58e7ca0ee..8d6716f36d220 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/plus-references.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/plus-references.py
@@ -1,19 +1,19 @@
 from dagster_cloud.metadata.source_code import link_code_references_to_git_if_cloud
 
-from dagster import Definitions, asset, with_source_code_references
+import dagster as dg
 
 
-@asset
+@dg.asset
 def my_asset(): ...
 
 
-@asset
+@dg.asset
 def another_asset(): ...
 
 
-defs = Definitions(
+defs = dg.Definitions(
     assets=link_code_references_to_git_if_cloud(
-        assets_defs=with_source_code_references([my_asset, another_asset]),
+        assets_defs=dg.with_source_code_references([my_asset, another_asset]),
         # This function also supports customizable path mapping, but usually
         # the defaults are fine.
     )
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/python-local-references.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/python-local-references.py
index a226637ddd086..37d1bc74c7a94 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/python-local-references.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/python-local-references.py
@@ -1,15 +1,15 @@
-from dagster import Definitions, asset, with_source_code_references
+import dagster as dg
 
 
-@asset
+@dg.asset
 def my_asset(): ...
 
 
-@asset
+@dg.asset
 def another_asset(): ...
 
 
-defs = Definitions(
+defs = dg.Definitions(
     # with_source_code_references() automatically attaches the proper metadata
-    assets=with_source_code_references([my_asset, another_asset])
+    assets=dg.with_source_code_references([my_asset, another_asset])
 )
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/runtime-metadata.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/runtime-metadata.py
index 51ea34c4b8b4c..974be8a84d2b2 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/runtime-metadata.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/runtime-metadata.py
@@ -1,12 +1,12 @@
-from dagster import MaterializeResult, MetadataValue, asset
+import dagster as dg
 
 
-@asset
+@dg.asset
 def my_asset():
     # Asset logic goes here
-    return MaterializeResult(
+    return dg.MaterializeResult(
         metadata={
             # file_size_kb will be rendered as a time series
-            "file_size_kb": MetadataValue.int(...)
+            "file_size_kb": dg.MetadataValue.int(...)
         }
     )
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/table-column-lineage-metadata.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/table-column-lineage-metadata.py
index 97abdbbae4c03..ed4a881624af1 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/table-column-lineage-metadata.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/table-column-lineage-metadata.py
@@ -1,13 +1,7 @@
-from dagster import (
-    AssetKey,
-    MaterializeResult,
-    TableColumnDep,
-    TableColumnLineage,
-    asset,
-)
+import dagster as dg
 
 
-@asset(deps=["source_bar", "source_baz"])
+@dg.asset(deps=["source_bar", "source_baz"])
 def my_asset():
     # The following TableColumnLineage models 3 tables:
     # source_bar, source_baz, and my_asset
@@ -17,23 +11,23 @@ def my_asset():
     #   - source_baz.column_baz
     # - my_asset.new_column_qux depends on:
     #   - source_bar.column_quuz
-    return MaterializeResult(
+    return dg.MaterializeResult(
         metadata={
-            "dagster/column_lineage": TableColumnLineage(
+            "dagster/column_lineage": dg.TableColumnLineage(
                 deps_by_column={
                     "new_column_foo": [
-                        TableColumnDep(
-                            asset_key=AssetKey("source_bar"),
+                        dg.TableColumnDep(
+                            asset_key=dg.AssetKey("source_bar"),
                             column_name="column_bar",
                         ),
-                        TableColumnDep(
-                            asset_key=AssetKey("source_baz"),
+                        dg.TableColumnDep(
+                            asset_key=dg.AssetKey("source_baz"),
                             column_name="column_baz",
                         ),
                     ],
                     "new_column_qux": [
-                        TableColumnDep(
-                            asset_key=AssetKey("source_bar"),
+                        dg.TableColumnDep(
+                            asset_key=dg.AssetKey("source_bar"),
                             column_name="column_quuz",
                         ),
                     ],
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/table-schema-metadata.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/table-schema-metadata.py
index b42aa49fbecaf..5775629bc33cf 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/table-schema-metadata.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/table-schema-metadata.py
@@ -1,19 +1,19 @@
-from dagster import MaterializeResult, TableColumn, TableSchema, asset
+import dagster as dg
 
 
 # Definition-time metadata
 # Here, we know the schema of the asset, so we can attach it to the asset decorator
-@asset(
+@dg.asset(
     deps=["source_bar", "source_baz"],
     metadata={
-        "dagster/column_schema": TableSchema(
+        "dagster/column_schema": dg.TableSchema(
             columns=[
-                TableColumn(
+                dg.TableColumn(
                     "name",
                     "string",
                     description="The name of the person",
                 ),
-                TableColumn(
+                dg.TableColumn(
                     "age",
                     "int",
                     description="The age of the person",
@@ -27,16 +27,16 @@ def my_asset(): ...
 
 # Runtime metadata
 # Here, the schema isn't known until the asset is materialized
-@asset(deps=["source_bar", "source_baz"])
+@dg.asset(deps=["source_bar", "source_baz"])
 def my_other_asset():
     column_names = ...
     column_types = ...
 
     columns = [
-        TableColumn(name, column_type)
+        dg.TableColumn(name, column_type)
         for name, column_type in zip(column_names, column_types)
     ]
 
-    return MaterializeResult(
-        metadata={"dagster/column_schema": TableSchema(columns=columns)}
+    return dg.MaterializeResult(
+        metadata={"dagster/column_schema": dg.TableSchema(columns=columns)}
     )
diff --git a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/tags.py b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/tags.py
index 036fe016e837e..81e92b991742e 100644
--- a/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/tags.py
+++ b/examples/docs_beta_snippets/docs_beta_snippets/guides/data-modeling/metadata/tags.py
@@ -1,12 +1,12 @@
-from dagster import AssetSpec, asset
+import dagster as dg
 
 
 # You can attach tags via the `tags` argument on the `@asset` decorator.
-@asset(tags={"domain": "marketing", "pii": "true"})
+@dg.asset(tags={"domain": "marketing", "pii": "true"})
 def my_asset(): ...
 
 
 # You can also use `tags` with `AssetSpec`
-my_external_asset = AssetSpec(
+my_external_asset = dg.AssetSpec(
     "my_external_asset", tags={"domain": "legal", "sensitive": ""}
 )