seanlindo · August 4, 2022 23:13 · Aug 4, 2022
diff --git a/example_io_manager.py b/example_io_manager.py
@@ -0,0 +1,17 @@
+from dagster import IOManager, io_manager
+
+
+class MyIOManager(IOManager):
+    def __init__(self):
+        self.storage_dict = {}
+
+    def handle_output(self, context, obj):
+        self.storage_dict[(context.step_key, context.name)] = obj
+
+    def load_input(self, context):
+        return self.storage_dict[(context.upstream_output.step_key, context.upstream_output.name)]
+
+
+@io_manager
+def my_io_manager(_):
+    return MyIOManager()
diff --git a/example_job.py b/example_job.py
@@ -0,0 +1,25 @@
+from dagster import asset, with_resources, repository, define_asset_job, SourceAsset
+from example_io_manager import my_io_manager
+
+raw_users = SourceAsset(key="raw_users", io_manager_key="test_io_manager")
+
+@asset(io_manager_key="test_io_manager")
+def upstream_asset(raw_users):
+    #TODO do some transformations
+    transformedUsers = raw_users
+
+    return transformedUsers
+
+#TODO what if there are many transformation steps? does each one have to be an asset? can they just be separate ops that work on the asset?
+
+@repository
+def repo():
+    return [
+        *with_resources(
+            [ raw_users, upstream_asset ],
+            resource_defs={
+                "test_io_manager": my_io_manager
+            },
+        ),
+        define_asset_job("process_users"),
+    ]
No results found