fix(oso_dagster): hotfix asset loading due to incorrect annotations (#4671)

Author: ravenac95

warehouse/oso_dagster/factories/common.py

@@ -1,9 +1,9 @@
 import inspect
-import logging
 import typing as t
 from dataclasses import dataclass, field
 from graphlib import TopologicalSorter
 
+import structlog
 from dagster import (
     AssetChecksDefinition,
     AssetsDefinition,
@@ -24,7 +24,7 @@
 type AssetKeyPrefixParam = dginternals.CoercibleToAssetKeyPrefix
 type FactoryJobDefinition = JobDefinition | dginternals.UnresolvedAssetJobDefinition
 
-logger = logging.getLogger(__name__)
+logger = structlog.get_logger(__name__)
 
 
 class GenericGCSAsset:
@@ -216,6 +216,12 @@ def run[T](
         annotations = f.__annotations__.copy()
         annotations.update(additional_annotations)
 
+        logger.debug(
+            f"running function {f.__name__} with annotations",
+            func_name=f.__name__,
+            annotations=annotations,
+        )
+
         resolved_resources: t.Dict[str, t.Any] = {}
         if ResourcesContext.resources_keyword_name in annotations:
             # If the function has a resources argument, we will pass the
@@ -308,6 +314,12 @@ def name(self) -> str:
         """Get the name of the asset factory."""
         return self._f.__name__
 
+    @property
+    def caller_filename(self) -> str:
+        if self._caller is None:
+            return "<unknown>"
+        return self._caller.filename
+
     @property
     def module(self) -> str:
         """Get the module of the asset factory."""

warehouse/oso_dagster/factories/dlt.py

@@ -1,4 +1,3 @@
-import functools
 import logging
 import typing as t
 from uuid import uuid4
@@ -90,7 +89,6 @@ def _decorator(
             asset_name = name or f.__name__
 
             @early_resources_asset_factory(caller_depth=2)
-            @functools.wraps(f)
             def _factory(
                 dlt_staging_destination: Destination,
                 dlt_warehouse_destination: Destination,

warehouse/oso_dagster/factories/loader.py

@@ -61,7 +61,9 @@ def load_all_assets_from_package(
 
     for module_info in pkgutil.walk_packages(package_path, package.__name__ + "."):
         module_name = module_info.name
-        with time_context(logger, f"loading module {module_name}"):
+        with time_context(
+            logger, f"loading module {module_name}", module_name=module_name
+        ):
             module = importlib.import_module(module_name)
             modules.append(module)
     factories = load_assets_factories_from_modules(modules, early_resources_dag)
@@ -75,6 +77,7 @@ def load_all_assets_from_package(
         with time_context(
             logger,
             f"generating assets for '{early_factory.name}'",
+            caller_filename=early_factory.caller_filename,
             loading_from_module=early_factory.module,
         ):
             resp = early_factory(resources, dependencies=resolved_deps)