Change to PermissiveConfig

nwp/assets/cams/cams_eu.py

@@ -27,13 +27,14 @@ def fetch_cams_eu_forecast_for_day(context: dagster.OpExecutionContext, config:
     c = cdsapi.Client()
 
     date: dt.datetime = dt.datetime.strptime(config.date, "%Y-%m-%d")
+    variables_to_use = config.dict().get("variables", VARIABLES)
 
     if date < dt.datetime.utcnow() - dt.timedelta(days=1095):
         raise ValueError('CAMS data is only available from 3 years ago onwards.')
 
     # Multi-level variables first
     for it in INIT_TIMES:
-        for var in VARIABLES:
+        for var in variables_to_use:
             fname: str = f'{config.raw_dir}/{date.strftime("%Y%m%d")}{it[:2]}_{var}.grib'
 
             c.retrieve(
@@ -103,7 +104,7 @@ def fetch_cams_eu_forecast_for_day(context: dagster.OpExecutionContext, config:
 
 
         # Validate that all files were downloaded
-        for var in VARIABLES:
+        for var in variables_to_use:
             fname: str = f'{config.raw_dir}/{date.strftime("%Y%m%d")}{it[:2]}_{var}.nc'
             if not os.path.isfile(fname):
                 raise FileNotFoundError(f"File {fname} was not downloaded.")

nwp/assets/cams/cams_global.py

@@ -158,11 +158,11 @@ def fetch_cams_forecast_for_day(context: dagster.OpExecutionContext, config: CAM
 
     # If date is more than 30 days ago, use slow variables, else use fast ones
     if (dt.datetime.now() - date).days > 30:
-        multi_variables: list[str] = SLOW_MULTI_VARIABLES
-        single_variables: list[str] = SLOW_SINGLE_VARIABLES
+        multi_variables: list[str] = config.dict().get("multi_variables", SLOW_MULTI_VARIABLES)
+        single_variables: list[str] = config.dict().get("single_variables", SLOW_SINGLE_VARIABLES)
     else:
-        multi_variables: list[str] = MULTI_LEVEL_NEW_VARIABLES
-        single_variables: list[str] = SINGLE_NEW_VARIABLES
+        multi_variables: list[str] = config.dict().get("multi_variables", MULTI_LEVEL_NEW_VARIABLES)
+        single_variables: list[str] = config.dict().get("single_variables", SINGLE_NEW_VARIABLES)
     # Multi-level variables first
     for it in INIT_TIMES:
         for var in multi_variables:

nwp/assets/cams/utils.py

@@ -1,6 +1,6 @@
 import dagster
 
 
-class CAMSConfig(dagster.Config):
+class CAMSConfig(dagster.PermissiveConfig):
     date: str
     raw_dir: str