Fix pyright type errors in project_multi_tenant example

- Cast row.prompt to str for pandas Scalar type in enriched_products
- Use Callable from collections.abc instead of builtin callable in io_managers
- Add ... body to Protocol methods in SupportsGenerate
- Use getattr loop instead of direct attribute access in test helper
- Add None guard before str `in` operator on skip_message

Made-with: Cursor

Author: C00ldudeNoonan

examples/project_multi_tenant/harbor_outfitters/assets/gold.py

@@ -16,7 +16,7 @@ def enriched_products(
     context.add_output_metadata(llm.runtime_metadata())
     rows: list[dict[str, object]] = []
     for row in catalog_prompt_inputs.itertuples(index=False):
-        response = llm.generate(row.prompt)
+        response = llm.generate(str(row.prompt))
         rows.append(
             {
                 "product_id": row.product_id,

examples/project_multi_tenant/shared/io_managers.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import time
+from collections.abc import Callable
 from pathlib import Path
 
 import dagster as dg
@@ -53,7 +54,7 @@ def _load_table(self, table_name: str) -> pd.DataFrame:
         with duckdb.connect(self.database) as connection:
             return connection.execute(f"select * from {self.db_schema}.{table_name}").fetch_df()
 
-    def _with_retry(self, fn: callable, *, action: str):
+    def _with_retry(self, fn: Callable, *, action: str):
         last_error: Exception | None = None
         for attempt in range(1, self.retry_attempts + 1):
             try:

examples/project_multi_tenant/shared/resources.py

@@ -15,9 +15,11 @@
 class SupportsGenerate(Protocol):
     def generate(self, prompt: str) -> str:
         """Generate a text response for the provided prompt."""
+        ...
 
     def runtime_metadata(self) -> dict[str, object]:
         """Return metadata describing the active model runtime."""
+        ...
 
 
 class BaseLLMResource(dg.ConfigurableResource):

examples/project_multi_tenant/tests/test_beacon_hq.py

@@ -58,11 +58,10 @@ def _build_success_context(
 def _get_run_success_event(
     result: dg.ExecuteInProcessResult,
 ) -> dg.DagsterEvent:
-    if hasattr(result, "get_run_success_event"):
-        return result.get_run_success_event()
-
-    if hasattr(result, "get_job_success_event"):
-        return result.get_job_success_event()
+    for method_name in ("get_run_success_event", "get_job_success_event"):
+        getter = getattr(result, method_name, None)
+        if getter is not None:
+            return getter()
 
     for event in reversed(result.all_events):
         if getattr(event, "event_type_value", "") == "PIPELINE_SUCCESS":
@@ -113,6 +112,7 @@ def test_beacon_sensor_waits_for_missing_upstream_job() -> None:
     response = beacon_after_upstream_success_sensor(_build_success_context(instance, harbor_result))
 
     assert isinstance(response, dg.SkipReason)
+    assert response.skip_message is not None
     assert "summit_risk_scoring_job" in response.skip_message