[dagster-airlift] [api-docs] mapping functions

dagster-io · Nov 7, 2024 · 1cb706e · 1cb706e
1 parent b627bfd
commit 1cb706e
Show file tree

Hide file tree

Showing 8 changed files with 59 additions and 39 deletions.
diff --git a/docs/content/api/modules.json.gz b/docs/content/api/modules.json.gz
diff --git a/docs/content/api/searchindex.json.gz b/docs/content/api/searchindex.json.gz
diff --git a/docs/content/api/sections.json.gz b/docs/content/api/sections.json.gz
diff --git a/docs/next/public/objects.inv b/docs/next/public/objects.inv
diff --git a/docs/sphinx/sections/api/apidocs/libraries/dagster-airlift.rst b/docs/sphinx/sections/api/apidocs/libraries/dagster-airlift.rst
@@ -17,28 +17,37 @@ AirflowInstance
 
 .. autoclass:: AirflowBasicAuthBackend 
 
-
 Assets & Definitions
 ^^^^^^^^^^^^^^^^^^^^
 
 .. autofunction:: build_defs_from_airflow_instance
 
+Mapping Dagster assets to Airflow tasks/dags:
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. autofunction:: assets_with_task_mappings
+
+.. autofunction:: assets_with_dag_mappings
+
+.. autofunction:: targeted_by_multiple_tasks
+
 Annotations for customizable components:
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 .. autoclass:: DagSelectorFn
 
 .. autoclass:: DagsterEventTransformerFn
 
+.. autoclass:: TaskHandleDict
+
 Objects for retrieving information about the Airflow/Dagster mapping:
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 .. autoclass:: DagInfo
 
 .. autoclass:: AirflowDefinitionsData
 
-
-
+.. currentmodule:: dagster_airlift.mwaa
 
 MWAA (dagster_airlift.mwaa)
 ---------------------------
@@ -47,3 +56,5 @@ MWAA (dagster_airlift.mwaa)
 .. autoclass:: MwaaSessionAuthBackend
 
 
+
+
diff --git a/examples/experimental/dagster-airlift/dagster_airlift/core/__init__.py b/examples/experimental/dagster-airlift/dagster_airlift/core/__init__.py
@@ -11,6 +11,10 @@
     build_airflow_mapped_defs as build_airflow_mapped_defs,
     build_defs_from_airflow_instance as build_defs_from_airflow_instance,
 )
+from .multiple_tasks import (
+    TaskHandleDict as TaskHandleDict,
+    targeted_by_multiple_tasks as targeted_by_multiple_tasks,
+)
 from .sensor.event_translation import (
     AssetEvent as AssetEvent,
     DagsterEventTransformerFn as DagsterEventTransformerFn,

diff --git a/examples/experimental/dagster-airlift/dagster_airlift/core/multiple_tasks.py b/examples/experimental/dagster-airlift/dagster_airlift/core/multiple_tasks.py
@@ -24,31 +24,32 @@ def targeted_by_multiple_tasks(
 
     .. code-block:: python
 
-    from dagster import Definitions, AssetSpec, asset
-    from dagster_airlift import build_defs_from_airflow_instance, dag_defs, task_defs, targeted_by_multiple_tasks
-
-    @asset
-    def scheduled_twice(): ...
-
-    defs = build_defs_from_airflow_instance(
-        airflow_instance=airflow_instance
-        defs=Definitions.merge(
-            dag_defs(
-                "other_dag",
-                task_defs(
-                    "task1",
-                    Definitions(assets=[other_asset]),
+        from dagster import Definitions, AssetSpec, asset
+        from dagster_airlift import build_defs_from_airflow_instance, dag_defs, task_defs, targeted_by_multiple_tasks
+
+        @asset
+        def scheduled_twice(): ...
+
+        defs = build_defs_from_airflow_instance(
+            airflow_instance=airflow_instance,
+            defs=Definitions.merge(
+                dag_defs(
+                    "other_dag",
+                    task_defs(
+                        "task1",
+                        Definitions(assets=[other_asset]),
+                    ),
                 ),
+                targeted_by_multiple_tasks(
+                    Definitions([scheduled_twice]),
+                    task_handles=[
+                        {"dag_id": "weekly_dag", "task_id": "task1"},
+                        {"dag_id": "daily_dag", "task_id": "task1"},
+                    ],
+                )
             ),
-            targeted_by_multiple_tasks(
-                Definitions([scheduled_twice]),
-                task_handles=[
-                    {"dag_id": "weekly_dag", "task_id": "task1"},
-                    {"dag_id": "daily_dag", "task_id": "task1"},
-                ],
-            )
-        ),
-    )
+        )
+
     """
     return replace_assets_in_defs(
         defs,

diff --git a/examples/experimental/dagster-airlift/dagster_airlift/core/top_level_dag_def_api.py b/examples/experimental/dagster-airlift/dagster_airlift/core/top_level_dag_def_api.py
@@ -78,22 +78,24 @@ def assets_with_task_mappings(
     this information. It is a list of dictionaries with keys "dag_id" and "task_id".
 
     Example:
-    .. code-block:: python
-        from dagster import AssetSpec, Definitions, asset
-        from dagster_airlift.core import assets_with_task_mappings
 
-        @asset
-        def asset_one() -> None: ...
+        .. code-block:: python
 
-        defs = Definitions(
-            assets=assets_with_task_mappings(
-                dag_id="dag_one",
-                task_mappings={
-                    "task_one": [asset_one],
-                    "task_two": [AssetSpec(key="asset_two"), AssetSpec(key="asset_three")],
-                },
+            from dagster import AssetSpec, Definitions, asset
+            from dagster_airlift.core import assets_with_task_mappings
+
+            @asset
+            def asset_one() -> None: ...
+
+            defs = Definitions(
+                assets=assets_with_task_mappings(
+                    dag_id="dag_one",
+                    task_mappings={
+                        "task_one": [asset_one],
+                        "task_two": [AssetSpec(key="asset_two"), AssetSpec(key="asset_three")],
+                    },
+                )
             )
-        )
     """
     assets_list = []
     for task_id, assets in task_mappings.items():
@@ -124,7 +126,9 @@ def assets_with_dag_mappings(
     this information. It is a list of strings, where each string is a dag_id which the asset is associated with.
 
     Example:
+
     .. code-block:: python
+
         from dagster import AssetSpec, Definitions, asset
         from dagster_airlift.core import assets_with_dag_mappings