Fix ruff

Author: cmpadden

python_modules/libraries/dagster-dlt/dagster_dlt/components/dlt_load_collection/component.py

@@ -81,8 +81,14 @@ class DltLoadSpecModel(Resolvable):
     translation: (
         Annotated[
             TranslationFn[DltResourceTranslatorData],
+            TranslationFnResolver[DltResourceTranslatorData](
+                lambda data: {"resource": data.resource, "pipeline": data.pipeline}
+            ),
+        ]
+        | None
+    ) = None
     partitions_def: Annotated[
-        Optional[PartitionsDefinition],
+        PartitionsDefinition | None,
         Resolver(
             resolve_partitions_def,
             model_field_type=HourlyPartitionsDefinitionModel
@@ -93,7 +99,7 @@ class DltLoadSpecModel(Resolvable):
         ),
     ] = None
     backfill_policy: Annotated[
-        Optional[BackfillPolicy],
+        BackfillPolicy | None,
         Resolver(
             resolve_backfill_policy,
             model_field_type=SingleRunBackfillPolicyModel | MultiRunBackfillPolicyModel,

python_modules/libraries/dagster-dlt/dagster_dlt_tests/test_dlt_load_collection_component.py

@@ -367,26 +367,41 @@ def get_asset_spec(self, data) -> AssetSpec:
     }
 
 
-def test_daily_partitioned_component(dlt_pipeline: Pipeline):
-    """Test that partitions_def and backfill_policy are correctly applied to assets."""
-    partitions_def = DailyPartitionsDefinition(start_date="2024-01-01")
-    backfill_policy = BackfillPolicy.single_run()
+def partitioned_load():
+    import dlt
 
-    context = ComponentTree.for_test().load_context
-    defs = DltLoadCollectionComponent(
-        loads=[
-            DltLoadSpecModel(
-                pipeline=dlt_pipeline,
-                source=dlt_source(),
-                partitions_def=partitions_def,
-                backfill_policy=backfill_policy,
-            )
-        ]
-    ).build_defs(context)
+    from dagster_dlt_tests.dlt_test_sources.duckdb_with_transformer import pipeline as dlt_source
 
-    assets_def = defs.resolve_assets_def(AssetKey(["example", "repos"]))
-    assert assets_def.partitions_def == partitions_def
-    assert assets_def.backfill_policy == backfill_policy
+    source = dlt_source()
+    pipeline = dlt.pipeline(
+        pipeline_name="pipeline",
+        destination="duckdb",
+        dataset_name="example",
+    )
+
+
+def test_daily_partitioned_component() -> None:
+    """Test that partitions_def and backfill_policy from YAML are applied to assets."""
+    with setup_dlt_component(
+        load_py_contents=partitioned_load,
+        defs_yaml_contents={
+            "type": "dagster_dlt.DltLoadCollectionComponent",
+            "attributes": {
+                "loads": [
+                    {
+                        "pipeline": ".load.pipeline",
+                        "source": ".load.source",
+                        "partitions_def": {"type": "daily", "start_date": "2024-01-01"},
+                        "backfill_policy": {"type": "single_run"},
+                    }
+                ]
+            },
+        },
+        setup_dlt_sources=lambda: None,
+    ) as (_component, defs):
+        assets_def = defs.resolve_assets_def(AssetKey(["example", "repos"]))
+        assert assets_def.partitions_def == DailyPartitionsDefinition(start_date="2024-01-01")
+        assert assets_def.backfill_policy == BackfillPolicy.single_run()
 
 
 def test_no_partitions_backwards_compatible(dlt_pipeline: Pipeline):