Mirroring commit 0036bba05bd7741a14ad7527fa89622f3626eb6b from internal

Author: elementl-devtools

dagster_open_platform/lib/__init__.py

@@ -0,0 +1,7 @@
+from dagster_open_platform.lib.executable_component import (
+    ExecutableComponent as ExecutableComponent,
+)
+from dagster_open_platform.lib.schedule import ScheduleComponent as ScheduleComponent
+from dagster_open_platform.lib.sling.cloud_product_ingest import (
+    ProdDbReplicationsComponent as ProdDbReplicationsComponent,
+)

dagster_open_platform/lib/executable_component.py

@@ -0,0 +1,99 @@
+import importlib
+import inspect
+from typing import Annotated, Callable, Literal, Optional
+
+import dagster as dg
+from dagster.components import (
+    Component,
+    ComponentLoadContext,
+    Model,
+    ResolutionContext,
+    Resolvable,
+    ResolvedAssetSpec,
+    Resolver,
+)
+from dagster_shared import check
+from typing_extensions import TypeAlias
+
+
+class DailyPartitionDefinitionModel(Resolvable, Model):
+    type: Literal["daily"] = "daily"
+    start_date: str
+    end_offset: int = 0
+
+
+def resolve_partition_definition(
+    context: ResolutionContext, model: DailyPartitionDefinitionModel
+) -> dg.PartitionsDefinition:
+    return dg.DailyPartitionsDefinition(
+        start_date=model.start_date,
+        end_offset=model.end_offset,
+    )
+
+
+ResolvedPartitionDefinition: TypeAlias = Annotated[
+    dg.DailyPartitionsDefinition,
+    Resolver(
+        resolve_partition_definition,
+        model_field_type=DailyPartitionDefinitionModel,
+        can_inject=True,
+    ),
+]
+
+
+def resolve_callable(context: ResolutionContext, model: str) -> Callable:
+    module_path, fn_name = model.rsplit(".", 1)
+    module = importlib.import_module(module_path)
+    return getattr(module, fn_name)
+
+
+ResolvableCallable: TypeAlias = Annotated[
+    Callable, Resolver(resolve_callable, model_field_type=str)
+]
+
+
+def get_resources_from_callable(func: Callable) -> list[str]:
+    sig = inspect.signature(func)
+    return [param.name for param in sig.parameters.values() if param.name != "context"]
+
+
+class ExecutableComponent(Component, Resolvable, Model):
+    """Executable Component represents an executable node in the asset graph.
+
+    It is comprised of an execute_fn, which is can be specified as a fully
+    resolved symbol reference in yaml. This makes it a plain ole' Python function
+    that does the execution within the asset graph.
+
+    You can pass an arbitrary number of assets or asset checks to the component.
+
+    With this structure this component replaces @asset, @multi_asset, @asset_check, and @multi_asset_check.
+    which can all be expressed as a single ExecutableComponent.
+    """
+
+    # inferred from the function name if not provided
+    name: Optional[str] = None
+    partitions_def: Optional[ResolvedPartitionDefinition] = None
+    assets: Optional[list[ResolvedAssetSpec]] = None
+    execute_fn: ResolvableCallable
+
+    def get_resource_keys(self) -> set[str]:
+        return set(get_resources_from_callable(self.execute_fn))
+
+    def build_defs(self, context: ComponentLoadContext) -> dg.Definitions:
+        required_resource_keys = self.get_resource_keys()
+
+        check.invariant(len(self.assets or []) > 0, "assets is required for now")
+
+        @dg.multi_asset(
+            name=self.name or self.execute_fn.__name__,
+            specs=self.assets,
+            partitions_def=self.partitions_def,
+            required_resource_keys=required_resource_keys,
+        )
+        def _assets_def(context: dg.AssetExecutionContext, **kwargs):
+            rd = context.resources.original_resource_dict
+            to_pass = {k: v for k, v in rd.items() if k in required_resource_keys}
+            check.invariant(set(to_pass.keys()) == required_resource_keys, "Resource keys mismatch")
+            return self.execute_fn(context, **to_pass)
+
+        return dg.Definitions(assets=[_assets_def])

dagster_open_platform_tests/test_gong/test_basic_execution.py

@@ -0,0 +1,67 @@
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pandas as pd
+from dagster import AssetsDefinition, DailyPartitionsDefinition, Definitions, materialize
+from dagster.components.core.context import use_component_load_context
+from dagster_open_platform.utils.components_test import build_component_and_defs
+from dagster_open_platform_tests.utils import component_context
+
+
+def test_basic_execute():
+    # need to import here to avoid double import
+    from dagster_open_platform.lib.executable_component import ExecutableComponent
+
+    ctx = component_context(Path("gong/component"))
+
+    with use_component_load_context(ctx):
+        component, defs = build_component_and_defs(ctx)
+
+        import dagster_open_platform.lib.executable_component
+
+        assert type(component).__name__ == "ExecutableComponent"
+        assert type(component) is dagster_open_platform.lib.executable_component.ExecutableComponent
+        assert isinstance(component, ExecutableComponent), f"Component is {type(component)}"
+
+        assert component.get_resource_keys() == {"claude", "snowflake", "slack_gong"}
+
+        defs = Definitions.merge(
+            defs,
+            Definitions(
+                resources=dict(
+                    snowflake=MagicMock(),
+                    claude=MagicMock(),
+                    slack_gong=MagicMock(),
+                )
+            ),
+        )
+
+        assets_def = defs.get_assets_def("gong_calls_transcript_ai")
+        assert isinstance(assets_def, AssetsDefinition)
+
+        assert isinstance(assets_def.partitions_def, DailyPartitionsDefinition)
+
+        mock_df = pd.DataFrame(
+            {
+                "CONVERSATION_ID": ["123", "456"],
+                "CALL_ANALYSIS_CONTEXT": [
+                    "Call between John and Jane discussing product features",
+                    "Call between Mike and Sarah about pricing",
+                ],
+            }
+        )
+
+        with patch(
+            "dagster_open_platform.defs.gong.component.gong_calls_transcript_ai.get_completed_calls_with_transcripts",
+            return_value=mock_df,
+        ):
+            with patch(
+                "dagster_open_platform.defs.gong.component.gong_calls_transcript_ai.create_call_analysis_context",
+                return_value=pd.DataFrame(),
+            ):
+                result = materialize(
+                    [assets_def],
+                    partition_key="2025-04-14",
+                )
+
+                assert result.success

pyproject.toml

@@ -54,6 +54,7 @@ dev = [
 ]
 
 tests = [
+    "mock>=5.2.0",
     "pytest",
     "responses",
     "syrupy",