fix(nwp): Typing for asset materializations (#149)

openclimatefix · Jan 27, 2025 · 566150b · 566150b
1 parent cc15b5b
commit 566150b
Show file tree

Hide file tree

Showing 7 changed files with 28 additions and 29 deletions.
diff --git a/pyproject.toml b/pyproject.toml
@@ -38,7 +38,7 @@ dev = [
     "unittest-xml-reporting",
     "dagit",
     # Linting
-    "ruff",
+    "ruff>=0.9.2",
     "types-pyyaml",
     "types-pytz",
     "types-requests",
@@ -57,6 +57,13 @@ enabled = true
 # Ruff configuration
 # * See https://beta.ruff.rs/docs/
 [tool.ruff]
+line-length = 100
+indent-width = 4
+# Use pydocstyle Google convention
+# See https://www.pydocstyle.org/en/stable/error_codes.html
+exclude = ["__init__.py"]
+
+[tool.ruff.lint]
 select = [
     "F",   # pyflakes
     "E",   # pycodestyle
@@ -77,26 +84,18 @@ select = [
     "D",   # pydocstyle
     "RUF", # ruff-specific rules
 ]
-line-length = 100
-indent-width = 4
-# Use pydocstyle Google convention
-# See https://www.pydocstyle.org/en/stable/error_codes.html
+fixable = ["ALL"]
 ignore = [
     "D203", "D213", "D215", "D400", "D401", "D404", "D406",
     "D407", "D408", "D409", "D413",
-    "ANN101",
 ]
-exclude = ["__init__.py"]
-fixable = ["ALL"]
+per-file-ignores = {"test*" = ["D", "ANN"]}
 
 [tool.ruff.format]
 quote-style = "double"
 indent-style = "space"
 line-ending = "auto"
 
-[tool.ruff.per-file-ignores]
-"test*" = ["D", "ANN"]
-
 # MyPy configuration
 # * See https://mypy.readthedocs.io/en/stable/index.html
 [tool.mypy]

diff --git a/src/dagster_dags/assets/nwp/ceda_mo_um_global.py b/src/dagster_dags/assets/nwp/ceda_mo_um_global.py
@@ -10,7 +10,7 @@
 """
 
 import os
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import dagster as dg
 from dagster_docker import PipesDockerClient
@@ -53,7 +53,7 @@
 def ceda_mo_um_global_asset(
     context: dg.AssetExecutionContext,
     pipes_docker_client: PipesDockerClient,
-) -> Any:  # noqa: ANN401
+    ) -> dg.MaterializeResult:
     """Dagster asset for MO Unified Model global NWP data from CEDA."""
     it: dt.datetime = context.partition_time_window.start
     return pipes_docker_client.run(
@@ -73,5 +73,5 @@ def ceda_mo_um_global_asset(
             "nano_cpus": 4e9,
         },
         context=context,
-    ).get_results()
+    ).get_materialize_result()
 
diff --git a/src/dagster_dags/assets/nwp/ecmwf_ens_stat_india.py b/src/dagster_dags/assets/nwp/ecmwf_ens_stat_india.py
@@ -11,7 +11,7 @@
 """
 
 import os
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import dagster as dg
 from dagster_docker import PipesDockerClient
@@ -55,7 +55,7 @@
 def ecmwf_ens_stat_india_asset(
     context: dg.AssetExecutionContext,
     pipes_docker_client: PipesDockerClient,
-) -> Any:  # noqa: ANN401
+    ) -> dg.MaterializeResult:
     """Dagster asset downloading ECMWF ENS data for India."""
     it: dt.datetime = context.partition_time_window.start
     return pipes_docker_client.run(
@@ -77,5 +77,5 @@ def ecmwf_ens_stat_india_asset(
             "nano_cpus": 4e9,
         },
         context=context,
-    ).get_results()
+    ).get_materialize_result()
 
diff --git a/src/dagster_dags/assets/nwp/ecmwf_hres_ifs_india.py b/src/dagster_dags/assets/nwp/ecmwf_hres_ifs_india.py
@@ -10,7 +10,7 @@
 """
 
 import os
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import dagster as dg
 from dagster_docker import PipesDockerClient
@@ -54,7 +54,7 @@
 def ecmwf_hres_ifs_india_asset(
     context: dg.AssetExecutionContext,
     pipes_docker_client: PipesDockerClient,
-) -> Any:  # noqa: ANN401
+    ) -> dg.MaterializeResult:
     """Dagster asset for HRES IFS model data covering India from ECMWF."""
     it: dt.datetime = context.partition_time_window.start
     return pipes_docker_client.run(
@@ -76,5 +76,5 @@ def ecmwf_hres_ifs_india_asset(
             "nano_cpus": 4e9,
         },
         context=context,
-    ).get_results()
+    ).get_materialize_result()
 
diff --git a/src/dagster_dags/assets/nwp/ecmwf_hres_ifs_west_europe.py b/src/dagster_dags/assets/nwp/ecmwf_hres_ifs_west_europe.py
@@ -10,7 +10,7 @@
 """
 
 import os
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import dagster as dg
 from dagster_docker import PipesDockerClient
@@ -54,7 +54,7 @@
 def ecmwf_hres_ifs_west_europe_asset(
     context: dg.AssetExecutionContext,
     pipes_docker_client: PipesDockerClient,
-) -> Any:  # noqa: ANN401
+) -> dg.MaterializeResult:
     """Dagster asset for HRES IFS NWP data covering Western Europe from ECMWF."""
     it: dt.datetime = context.partition_time_window.start
     return pipes_docker_client.run(
@@ -76,5 +76,5 @@ def ecmwf_hres_ifs_west_europe_asset(
             "nano_cpus": 4e9,
         },
         context=context,
-    ).get_results()
+    ).get_materialize_result()
 
diff --git a/src/dagster_dags/assets/nwp/noaa-gfs-global.py b/src/dagster_dags/assets/nwp/noaa-gfs-global.py
@@ -11,7 +11,7 @@
 """
 
 import os
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import dagster as dg
 from dagster_docker import PipesDockerClient
@@ -54,7 +54,7 @@
 def ncep_gfs_global_asset(
     context: dg.AssetExecutionContext,
     pipes_docker_client: PipesDockerClient,
-) -> Any:  # noqa: ANN401
+    ) -> dg.MaterializeResult:
     """Dagster asset for NCEP GFS global forecast model data."""
     it: dt.datetime = context.partition_time_window.start
     return pipes_docker_client.run(
@@ -72,5 +72,5 @@ def ncep_gfs_global_asset(
             "nano_cpus": 4e9,
         },
         context=context,
-    ).get_results()
+    ).get_materialize_result()
 
diff --git a/src/dagster_dags/assets/sat/eumetsat_iodc_lrv.py b/src/dagster_dags/assets/sat/eumetsat_iodc_lrv.py
@@ -11,7 +11,7 @@
 """
 
 import os
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import dagster as dg
 from dagster_docker import PipesDockerClient
@@ -48,7 +48,7 @@
 def eumetsat_seviri_lrv_asset(
     context: dg.AssetExecutionContext,
     pipes_docker_client: PipesDockerClient,
-) -> Any: # noqa: ANN401
+) -> dg.MaterializeResult:
     """Dagster asset for EUMETSAT's RSS service, low resolution."""
     it: dt.datetime = context.partition_time_window.start
     return pipes_docker_client.run(
@@ -58,5 +58,5 @@ def eumetsat_seviri_lrv_asset(
             "volumes": [f"{ARCHIVE_FOLDER}:/work"],
         },
         context=context,
-    ).get_results()
+    ).get_materialize_result()