fix(run_launcher): bump scaleway sdk with breaking changes (#8)

* fix(run_launcher): bump scaleway sdk with breaking changes
* fix: ruff fmt
* fix: bump scaleway sdk to 2.1.0

Author: cyclimse

dagster_scaleway/serverless_job_context.py

@@ -57,7 +57,7 @@ class ScalewayServerlessJobContext(
         [
             ("docker_image", Optional[str]),
             ("env_vars", Sequence[str]),
-            ("region", str),
+            ("region", Optional[str]),
             ("memory_limit", int),
             ("cpu_limit", int),
         ],

dagster_scaleway/serverless_job_launcher.py

@@ -4,15 +4,14 @@
 from dagster._core.launcher.base import (
     CheckRunHealthResult,
     LaunchRunContext,
-    ResumeRunContext,
     RunLauncher,
     WorkerStatus,
 )
 from dagster._core.origin import JobPythonOrigin
 from dagster._core.storage.dagster_run import DagsterRun
 from dagster._core.storage.tags import DOCKER_IMAGE_TAG
 from dagster._core.utils import parse_env_var
-from dagster._grpc.types import ExecuteRunArgs, ResumeRunArgs
+from dagster._grpc.types import ExecuteRunArgs
 from dagster._serdes import ConfigurableClass
 from dagster._serdes.config_class import ConfigurableClassData
 from typing_extensions import Self
@@ -115,64 +114,39 @@ def _get_semantic_job_name(self, dagster_run: DagsterRun) -> str:
             return list(dagster_run.asset_selection)[-1].to_user_string()
         return dagster_run.run_id
 
-    def _create_or_update_job_definition(
+    def _get_or_create_job_definition(
         self,
         client: scaleway.Client,
         run: DagsterRun,
         docker_image: str,
-        command: list[str],
     ) -> scw.JobDefinition:
         serverless_job_context = self.get_serverless_job_context(run)
         api = scw.JobsV1Alpha1API(client)
 
-        job_def_env = dict(
-            [parse_env_var(env_var) for env_var in serverless_job_context.env_vars]
-        )
-        job_def_env["DAGSTER_RUN_JOB_NAME"] = run.job_name
-        job_def_env["DAGSTER_RUN_ID"] = run.run_id
-        job_def_env["INPUT_JSON"] = command[-1]
-
-        api = scw.JobsV1Alpha1API(client)
-
-        wrapped_command = [COMMAND_WRAPPER] + command[:-1]
-        description = (
-            f"JobDefinition for {run.job_name}."
-            + " "
-            + "Created by the ServerlessJobRunLauncher from dagster-scaleway."
-        )
         job_def_name = self._get_semantic_job_name(run)
 
         for job_def in api.list_job_definitions_all():
             if job_def.name != job_def_name:
                 continue
 
-            job_def = api.update_job_definition(
-                job_definition_id=job_def.id,
-                name=job_def_name,
-                image_uri=docker_image,
-                environment_variables=job_def_env,
-                command=" ".join(wrapped_command),
-                memory_limit=serverless_job_context.memory_limit,
-                cpu_limit=serverless_job_context.cpu_limit,
-                description=description,
+            return self._update_job_definition(
+                client, serverless_job_context, job_def, docker_image
             )
 
-            self._instance.report_engine_event(
-                message=f"Updated job {job_def.id} for Dagster run {run.run_id}",
-                dagster_run=run,
-                cls=self.__class__,
-            )
+        job_def_env = dict(
+            [parse_env_var(env_var) for env_var in serverless_job_context.env_vars]
+        )
 
-            return job_def
+        job_def_name = self._get_semantic_job_name(run)
 
         job_def = api.create_job_definition(
             name=job_def_name,
             image_uri=docker_image,
             environment_variables=job_def_env,
-            command=" ".join(wrapped_command),
+            command=COMMAND_WRAPPER,
             memory_limit=serverless_job_context.memory_limit,
             cpu_limit=serverless_job_context.cpu_limit,
-            description=description,
+            description="Created by the ServerlessJobRunLauncher from dagster-scaleway.",
             project_id=client.default_project_id,
         )
 
@@ -184,21 +158,70 @@ def _create_or_update_job_definition(
 
         return job_def
 
-    def _launch_serverless_job_with_command(
+    def _update_job_definition(
+        self,
+        client: scaleway.Client,
+        context: ScalewayServerlessJobContext,
+        job_def: scw.JobDefinition,
+        docker_image: str,
+    ) -> scw.JobDefinition:
+        """Reconcile the job definition with the desired state"""
+        api = scw.JobsV1Alpha1API(client)
+        has_changed = False
+
+        if job_def.image_uri != docker_image:
+            job_def.image_uri = docker_image
+            has_changed = True
+
+        if job_def.memory_limit != context.memory_limit:
+            job_def.memory_limit = context.memory_limit
+            has_changed = True
+
+        if job_def.cpu_limit != context.cpu_limit:
+            job_def.cpu_limit = context.cpu_limit
+            has_changed = True
+
+        if not has_changed:
+            return job_def
+
+        return api.update_job_definition(
+            job_definition_id=job_def.id,
+            image_uri=docker_image,
+            memory_limit=context.memory_limit,
+            cpu_limit=context.cpu_limit,
+        )
+
+    def _start_serverless_job_with_command(
         self, run: DagsterRun, docker_image: str, command: list[str]
     ):
         serverless_job_context = self.get_serverless_job_context(run)
         client = self._get_client(serverless_job_context)
         api = scw.JobsV1Alpha1API(client)
 
-        job_def = self._create_or_update_job_definition(
-            client, run, docker_image, command
+        job_def = self._get_or_create_job_definition(client, run, docker_image)
+
+        extra_env = {
+            "DAGSTER_RUN_JOB_NAME": run.job_name,
+            "DAGSTER_RUN_ID": run.run_id,
+            "INPUT_JSON": command[-1],
+        }
+        wrapped_command = [COMMAND_WRAPPER] + command[:-1]
+
+        start_response = api.start_job_definition(
+            job_definition_id=job_def.id,
+            command=" ".join(wrapped_command),
+            environment_variables=extra_env,
         )
 
-        job_run = api.start_job_definition(job_definition_id=job_def.id)
+        if len(start_response.job_runs) != 1:
+            raise RuntimeError(
+                f"Expected 1 job run to be created, got {len(start_response.job_runs)}"
+            )
+
+        job_run = start_response.job_runs[0]
 
         self._instance.report_engine_event(
-            message=f"Started job definition {job_def.name} with run id {job_run.id} for Dagster run {run.run_id}",
+            message=f"Started job definition {job_def.name} with job run id {job_run.id} for Dagster run {run.run_id}",
             dagster_run=run,
             cls=self.__class__,
         )
@@ -223,25 +246,11 @@ def launch_run(self, context: LaunchRunContext) -> None:
             instance_ref=self._instance.get_ref(),
         ).get_command_args()
 
-        self._launch_serverless_job_with_command(run, docker_image, command)
+        self._start_serverless_job_with_command(run, docker_image, list(command))
 
     @property
     def supports_resume_run(self):
-        # TODO?: check if we can resume a run
-        return True
-
-    def resume_run(self, context: ResumeRunContext) -> None:
-        run = context.dagster_run
-        job_code_origin = check.not_none(context.job_code_origin)
-        docker_image = self._get_docker_image(job_code_origin)
-
-        command = ResumeRunArgs(
-            job_origin=job_code_origin,
-            run_id=run.run_id,
-            instance_ref=self._instance.get_ref(),
-        ).get_command_args()
-
-        self._launch_serverless_job_with_command(run, docker_image, command)
+        return False
 
     def _get_scaleway_job_run_from_dagster_run(self, run) -> Optional[scw.JobRun]:
         if not run or run.is_finished:
@@ -303,18 +312,18 @@ def check_run_worker_health(self, run: DagsterRun):
         job_run = self._get_scaleway_job_run_from_dagster_run(run)
         if job_run is None:
             return CheckRunHealthResult(
-                WorkerStatus.NOT_FOUND,
+                status=WorkerStatus.NOT_FOUND,
                 run_worker_id=run.run_id,
                 msg=f"Unable to find Scaleway job run with id {run.run_id} for Dagster run {run.run_id}",
             )
 
-        health = CheckRunHealthResult(run_worker_id=run.run_id)
-        health.transient = job_run.state in scw.JOB_RUN_TRANSIENT_STATUSES
-        health.status = SERVERLESS_JOBS_STATES_TO_WORKER_STATUS.get(
-            job_run.state, WorkerStatus.UNKNOWN
+        health = CheckRunHealthResult(
+            status=SERVERLESS_JOBS_STATES_TO_WORKER_STATUS.get(
+                job_run.state, WorkerStatus.UNKNOWN
+            ),
+            run_worker_id=run.run_id,
+            msg=job_run.error_message,
+            transient=job_run.state in scw.JOB_RUN_TRANSIENT_STATUSES,
         )
 
-        if job_run.error_message:
-            health.msg = job_run.error_message
-
         return health