fix(repo): Use just one code location

Containerfile

@@ -13,12 +13,11 @@ COPY pyproject.toml /opt/dagster/app
 # DagsterInstance.
 RUN uv sync
 
-# Set the code location module to be loaded by the gRPC server
-ENV MODULE_NAME=local_archives
+EXPOSE 4000
 
 # Using CMD rather than RUN allows the command to be overridden in
 # run launchers or executors to run other commands using this image.
 # This is important as runs are executed inside this container.
 ENTRYPOINT ["uv", "run"]
-CMD ["dagster", "api", "grpc", "-h", "0.0.0.0", "-p", "4000", "-m", "local_archives"]
+CMD ["dagster", "api", "grpc", "-h", "0.0.0.0", "-p", "4000", "-m", "dagster_dags"]
 

infrastructure/docker-compose.yaml

@@ -25,8 +25,6 @@ services:
       PGDATA: "/var/lib/postgresql/data"
     volumes:
       - dagster-pgdata-vol:/var/lib/postgresql/data
-    expose:
-      - "5432" # Publishes on network but not to host
     networks: ["dagster-network"]
     healthcheck:
       test: pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB
@@ -37,16 +35,14 @@ services:
   # This service runs the gRPC server that loads user code, used by both dagster-webserver
   # and dagster-daemon. By setting DAGSTER_CURRENT_IMAGE to its own image, we tell the
   # run launcher to use this same image when launching runs in a new container as well.
-  dagster-codeserver_local-archives:
-    container_name: dagster-codeserver_local-archives
+  dagster-codeserver:
+    container_name: dagster-codeserver
     image: ghcr.io/openclimatefix/dagster-dags:devsjc-code-container
     restart: always
     environment:
       <<: *postgres-variables
       DAGSTER_CURRENT_IMAGE: "ghcr.io/openclimatefix/dagster-dags"
       DAGSTER_HOME: "/opt/dagster/home"
-    expose:
-      - "4000"
     configs: *dagster-configs
     networks: ["dagster-network"]
 
@@ -111,9 +107,9 @@ configs:
     content: |
       load_from:
         - grpc_server:
-            host: "dagster-codeserver_local-archives"
+            host: "dagster-codeserver"
             port: 4000
-            location_name: "local_archives"
+            location_name: "dagster_dags"
 
   dagster.yaml:
     content: |

src/dagster_dags/__init__.py

@@ -0,0 +1,4 @@
+from .definitions import defs
+
+__all__ = ['defs']
+

src/dagster_dags/definitions.py

@@ -0,0 +1,48 @@
+import dagster as dg
+from dagster_docker import PipesDockerClient
+
+from . import nwp, pv, sat
+
+
+class BaseStorageResource(dg.ConfigurableResource):
+    """A resource defining where to store external data.
+
+    Consists of a location on disk accessible to the Dagster instance.
+    Where possible all data not handled by Dagster's IO managers (e.g.
+    data downloaded via dagster pipes docker instances) will use this
+    path as a base location.
+    """
+
+    location: str
+
+nwp_assets = dg.load_assets_from_package_module(
+    package_module=nwp,
+    group_name="nwp",
+    key_prefix="nwp",
+)
+
+sat_assets = dg.load_assets_from_package_module(
+    package_module=sat,
+    group_name="sat",
+    key_prefix="sat",
+)
+
+pv_assets = dg.load_assets_from_package_module(
+    package_module=pv,
+    group_name="pv",
+    key_prefix="pv",
+)
+
+defs = dg.Definitions(
+    assets=[*nwp_assets, *sat_assets, *pv_assets],
+    resources={
+        "pipes_subprocess_client": dg.PipesSubprocessClient(),
+        "pipes_docker_client": PipesDockerClient(),
+        "base_storage_resource": BaseStorageResource(
+            location=dg.EnvVar("BASE_STORAGE_PATH"),
+        ),
+    },
+    jobs=[],
+    schedules=[],
+)
+

tests/__init__.py

@@ -0,0 +1 @@
+

tests/compile_test.py

@@ -2,14 +2,15 @@
 
 import dagster as dg
 
-from src.local_archives import nwp_assets, sat_assets
+from src.dagster_dags import defs
 
 
 class TestAssetKeyPrefixes(unittest.TestCase):
     def test_nwp_asset_key_prefixes(self) -> None:
         """Test asset keys for all nwp assets have the correct key structure."""
-        for asset in [*nwp_assets, *sat_assets]:
-            if isinstance(asset, dg.AssetsDefinition):
-                # Ensure that the prefix is one of the expected flavours
-                self.assertIn( asset.key.path[0], ["nwp", "sat", "air"])
+        if defs.assets is not None:
+            for asset in defs.assets:
+                if isinstance(asset, dg.AssetsDefinition):
+                    # Ensure that the prefix is one of the expected flavours
+                    self.assertIn( asset.key.path[0], ["nwp", "sat", "air", "pv"])
 

tests/cloud_archives/pv/test_passiv.py → tests/test_passiv.py

@@ -1,7 +1,7 @@
 import unittest
 from datetime import UTC, datetime
 
-from src.cloud_archives.pv.passiv.passiv_monthly import get_monthly_passiv_data
+from src.dagster_dags.pv.passiv.passiv_monthly import get_monthly_passiv_data
 
 
 @unittest.skip("rawdata endpoint not on new URL")