From d0c5855c2ed08def8582e26ce15f164e360cf3b9 Mon Sep 17 00:00:00 2001 From: Elliott Whiting Date: Tue, 30 Jun 2026 16:54:45 +0100 Subject: [PATCH 1/4] Add scheduled job functions --- domino/domino.py | 412 +++++++++++++++++++++++++++-------- domino/routes.py | 7 + tests/test_scheduled_jobs.py | 217 ++++++++++++++++++ 3 files changed, 550 insertions(+), 86 deletions(-) create mode 100644 tests/test_scheduled_jobs.py diff --git a/domino/domino.py b/domino/domino.py index 3ff8dce5..02d1d7d8 100644 --- a/domino/domino.py +++ b/domino/domino.py @@ -566,80 +566,6 @@ def validate_is_on_demand_spark_supported(): f"Minimum support version {MINIMUM_ON_DEMAND_SPARK_CLUSTER_SUPPORT_DOMINO_VERSION}" ) - def validate_distributed_compute_cluster_properties(): - if not helpers.is_compute_cluster_properties_supported(self._version): - raise exceptions.UnsupportedFieldException( - f"'compute_cluster_properties' is not supported in Domino {self._version}." - ) - - required_keys = [ - "clusterType", - "computeEnvironmentId", - "masterHardwareTierId", - "workerHardwareTierId", - "workerCount", - ] - required_key_overrides = {("masterHardwareTierId", "MPI"): False} - - for key in required_keys: - key_required = required_key_overrides.get( - (key, compute_cluster_properties["clusterType"]), True - ) - - if key_required and (key not in compute_cluster_properties): - raise exceptions.MissingRequiredFieldException( - f"{key} is required in compute_cluster_properties" - ) - - if not helpers.is_cluster_type_supported( - self._version, compute_cluster_properties["clusterType"] - ): - supported_types = [ - ct - for ct, min_version in CLUSTER_TYPE_MIN_SUPPORT - if helpers.is_cluster_type_supported(self._version, ct) - ] - supported_types_str = ", ".join(supported_types) - raise exceptions.MalformedInputException( - f"Domino {self._version} does not support cluster type {compute_cluster_properties['clusterType']}." - + f" This version of Domino supports the following cluster types: {supported_types_str}" - ) - - def throw_if_information_invalid(key: str, info: dict) -> None: - try: - self._validate_information_data_type(info) - except Exception as e: - raise exceptions.MalformedInputException( - f"{key} in compute_cluster_properties failed validation: {e}" - ) - - if "workerStorage" in compute_cluster_properties: - throw_if_information_invalid( - "workerStorage", compute_cluster_properties["workerStorage"] - ) - - if compute_cluster_properties["workerCount"] < 1: - raise exceptions.MalformedInputException( - "compute_cluster_properties.workerCount must be greater than 0" - ) - - if ( - "maxWorkerCount" in compute_cluster_properties - and not helpers.is_compute_cluster_autoscaling_supported(self._version) - ): - raise exceptions.UnsupportedFieldException( - f"'maxWorkerCount' is not supported in Domino {self._version}." - ) - - if "masterHardwareTierId" in compute_cluster_properties: - self._validate_hardware_tier_id( - compute_cluster_properties["masterHardwareTierId"] - ) - - self._validate_hardware_tier_id( - compute_cluster_properties["workerHardwareTierId"] - ) - def validate_is_external_volume_mounts_supported(): if not helpers.is_external_volume_mounts_supported(self._version): raise exceptions.ExternalVolumeMountsNotSupportedException( @@ -661,18 +587,9 @@ def validate_is_external_volume_mounts_supported(): if external_volume_mounts is not None: validate_is_external_volume_mounts_supported() if compute_cluster_properties is not None: - validate_distributed_compute_cluster_properties() - - validated_compute_cluster_properties = compute_cluster_properties.copy() - - if "masterHardwareTierId" in compute_cluster_properties: - validated_compute_cluster_properties["masterHardwareTierId"] = { - "value": compute_cluster_properties["masterHardwareTierId"] - } - - validated_compute_cluster_properties["workerHardwareTierId"] = { - "value": compute_cluster_properties["workerHardwareTierId"] - } + validated_compute_cluster_properties = self._validate_compute_cluster_properties( + compute_cluster_properties + ) elif on_demand_spark_cluster_properties is not None: validate_is_on_demand_spark_supported() @@ -857,6 +774,249 @@ def get_job_status(job_identifier): self.process_log(stdout_msg) return job_status + def scheduled_jobs_list(self, project_id: Optional[str] = None) -> list: + """ + List all scheduled jobs for a project. + :param project_id: The project ID (defaults to the current project) + :return: List of scheduled job definitions + """ + project_id = project_id or self.project_id + url = self._routes.scheduled_jobs(project_id) + return self._get(url) + + def scheduled_job_get(self, scheduled_job_id: str) -> dict: + """ + Retrieve a scheduled job by ID. + :param scheduled_job_id: The scheduled job ID + :return: Scheduled job definition + """ + url = self._routes.scheduled_job(self.project_id, scheduled_job_id) + return self.request_manager.get(url).json() + + def scheduled_job_create( + self, + title: str, + command: str, + cron_string: str, + timezone_id: str, + hardware_tier_identifier: str, + environment_revision_spec: Union[str, dict] = "ActiveRevision", + is_paused: bool = False, + allow_concurrent_execution: bool = False, + is_custom_schedule: bool = True, + notify_on_complete_email_addresses: Optional[List[str]] = None, + publish_model_id: Optional[str] = None, + capacity_type: Optional[str] = None, + override_environment_id: Optional[str] = None, + dataset_config: Optional[str] = None, + compute_cluster_properties: Optional[dict] = None, + external_volume_mounts: Optional[List[str]] = None, + snapshot_datasets_on_completion: Optional[bool] = None, + snapshot_net_app_volumes_on_completion: Optional[bool] = None, + main_repo_git_ref: Optional[dict] = None, + ) -> dict: + """ + Create a new scheduled job. + :param title: string + Display name for the scheduled job + :param command: string + Command to run (e.g. "main.py arg1 arg2") + :param cron_string: string + Cron expression defining the schedule (e.g. "0 9 * * 1") + :param timezone_id: string + IANA timezone (e.g. "America/New_York", "UTC") + :param hardware_tier_identifier: string + Hardware tier identifier to run on + :param environment_revision_spec: string or dict (Optional) + One of "ActiveRevision", "LatestRevision", + "RestrictedRevision", or {"revisionId": ""} + :param is_paused: bool (Optional) + Whether to create the job in a paused state (default False) + :param allow_concurrent_execution: bool (Optional) + Allow multiple instances to run simultaneously (default False) + :param is_custom_schedule: bool (Optional) + Whether the cron string is a custom schedule (default True) + :param notify_on_complete_email_addresses: list of string (Optional) + Email addresses to notify on completion + :param publish_model_id: string (Optional) + Model ID to publish results to + :param capacity_type: string (Optional) + "on-demand" or "spot" + :param override_environment_id: string (Optional) + Override the project's default environment + :param dataset_config: string (Optional) + Dataset configuration + :param compute_cluster_properties: dict (Optional) + Compute cluster configuration (Spark/Ray/Dask/MPI) + :param external_volume_mounts: list of string (Optional) + External volume mount IDs + :param snapshot_datasets_on_completion: bool (Optional) + Snapshot datasets when the job completes + :param snapshot_net_app_volumes_on_completion: bool (Optional) + Snapshot NetApp volumes when the job completes + :param main_repo_git_ref: dict (Optional) + Git ref for the main repo, e.g. {"type": "branches", "value": "main"} + :return: Created scheduled job definition + """ + self._validate_hardware_tier_id(hardware_tier_identifier) + if override_environment_id is not None: + self._validate_environment_id(override_environment_id) + if compute_cluster_properties is not None: + compute_cluster_properties = self._validate_compute_cluster_properties( + compute_cluster_properties + ) + scheduled_by_user_id = self.get_user_id(self._owner_username) + request: Dict[str, Any] = { + "title": title, + "command": command, + "schedule": {"cronString": cron_string, "isCustom": is_custom_schedule}, + "timezoneId": timezone_id, + "isPaused": is_paused, + "allowConcurrentExecution": allow_concurrent_execution, + "hardwareTierIdentifier": hardware_tier_identifier, + "environmentRevisionSpec": environment_revision_spec, + "scheduledByUserId": scheduled_by_user_id, + "notifyOnCompleteEmailAddresses": notify_on_complete_email_addresses or [], + } + if publish_model_id is not None: + request["publishModelId"] = publish_model_id + if capacity_type is not None: + request["capacityType"] = capacity_type + if override_environment_id is not None: + request["overrideEnvironmentId"] = override_environment_id + if dataset_config is not None: + request["datasetConfig"] = dataset_config + if compute_cluster_properties is not None: + request["computeClusterProperties"] = compute_cluster_properties + if external_volume_mounts is not None: + request["externalVolumeMounts"] = external_volume_mounts + if snapshot_datasets_on_completion is not None: + request["snapshotDatasetsOnCompletion"] = snapshot_datasets_on_completion + if snapshot_net_app_volumes_on_completion is not None: + request["snapshotNetAppVolumesOnCompletion"] = snapshot_net_app_volumes_on_completion + if main_repo_git_ref is not None: + request["mainRepoGitRef"] = main_repo_git_ref + url = self._routes.scheduled_jobs(self.project_id) + return self.request_manager.post(url, json=request).json() + + def scheduled_job_update( + self, + scheduled_job_id: str, + title: str, + command: str, + cron_string: str, + timezone_id: str, + hardware_tier_identifier: str, + environment_revision_spec: Union[str, dict] = "ActiveRevision", + is_paused: bool = False, + allow_concurrent_execution: bool = False, + is_custom_schedule: bool = True, + notify_on_complete_email_addresses: Optional[List[str]] = None, + publish_model_id: Optional[str] = None, + capacity_type: Optional[str] = None, + override_environment_id: Optional[str] = None, + dataset_config: Optional[str] = None, + compute_cluster_properties: Optional[dict] = None, + external_volume_mounts: Optional[List[str]] = None, + snapshot_datasets_on_completion: Optional[bool] = None, + snapshot_net_app_volumes_on_completion: Optional[bool] = None, + main_repo_git_ref: Optional[dict] = None, + ) -> dict: + """ + Update an existing scheduled job. All scheduling and execution parameters are replaced. + :param scheduled_job_id: string + The scheduled job ID to update + :param title: string + Display name for the scheduled job + :param command: string + Command to run (e.g. "main.py arg1 arg2") + :param cron_string: string + Cron expression defining the schedule (e.g. "0 9 * * 1") + :param timezone_id: string + IANA timezone (e.g. "America/New_York", "UTC") + :param hardware_tier_identifier: string + Hardware tier identifier to run on + :param environment_revision_spec: string or dict (Optional) + One of "ActiveRevision", "LatestRevision", + "RestrictedRevision", or {"revisionId": ""} + :param is_paused: bool (Optional) + Whether the job is paused (default False) + :param allow_concurrent_execution: bool (Optional) + Allow multiple instances to run simultaneously (default False) + :param is_custom_schedule: bool (Optional) + Whether the cron string is a custom schedule (default True) + :param notify_on_complete_email_addresses: list of string (Optional) + Email addresses to notify on completion + :param publish_model_id: string (Optional) + Model ID to publish results to + :param capacity_type: string (Optional) + "on-demand" or "spot" + :param override_environment_id: string (Optional) + Override the project's default environment + :param dataset_config: string (Optional) + Dataset configuration + :param compute_cluster_properties: dict (Optional) + Compute cluster configuration (Spark/Ray/Dask/MPI) + :param external_volume_mounts: list of string (Optional) + External volume mount IDs + :param snapshot_datasets_on_completion: bool (Optional) + Snapshot datasets when the job completes + :param snapshot_net_app_volumes_on_completion: bool (Optional) + Snapshot NetApp volumes when the job completes + :param main_repo_git_ref: dict (Optional) + Git ref for the main repo, e.g. {"type": "branches", "value": "main"} + :return: Updated scheduled job definition + """ + self._validate_hardware_tier_id(hardware_tier_identifier) + if override_environment_id is not None: + self._validate_environment_id(override_environment_id) + if compute_cluster_properties is not None: + compute_cluster_properties = self._validate_compute_cluster_properties( + compute_cluster_properties + ) + scheduled_by_user_id = self.get_user_id(self._owner_username) + request: Dict[str, Any] = { + "title": title, + "command": command, + "schedule": {"cronString": cron_string, "isCustom": is_custom_schedule}, + "timezoneId": timezone_id, + "isPaused": is_paused, + "allowConcurrentExecution": allow_concurrent_execution, + "hardwareTierIdentifier": hardware_tier_identifier, + "environmentRevisionSpec": environment_revision_spec, + "scheduledByUserId": scheduled_by_user_id, + "notifyOnCompleteEmailAddresses": notify_on_complete_email_addresses or [], + } + if publish_model_id is not None: + request["publishModelId"] = publish_model_id + if capacity_type is not None: + request["capacityType"] = capacity_type + if override_environment_id is not None: + request["overrideEnvironmentId"] = override_environment_id + if dataset_config is not None: + request["datasetConfig"] = dataset_config + if compute_cluster_properties is not None: + request["computeClusterProperties"] = compute_cluster_properties + if external_volume_mounts is not None: + request["externalVolumeMounts"] = external_volume_mounts + if snapshot_datasets_on_completion is not None: + request["snapshotDatasetsOnCompletion"] = snapshot_datasets_on_completion + if snapshot_net_app_volumes_on_completion is not None: + request["snapshotNetAppVolumesOnCompletion"] = snapshot_net_app_volumes_on_completion + if main_repo_git_ref is not None: + request["mainRepoGitRef"] = main_repo_git_ref + url = self._routes.scheduled_job(self.project_id, scheduled_job_id) + return self.request_manager.put(url, json=request).json() + + def scheduled_job_delete(self, scheduled_job_id: str): + """ + Delete a scheduled job. + :param scheduled_job_id: The scheduled job ID to delete + :return: Response from the server + """ + url = self._routes.scheduled_job(self.project_id, scheduled_job_id) + return self.request_manager.delete(url) + def files_list(self, commit_id=_UNSET, path="/", **kwargs): commit_id = _resolve_renamed_kwarg( commit_id, "commitId", "commit_id", kwargs, None @@ -2096,6 +2256,86 @@ def _useable_environments_list(self): "environments" ] + def _validate_compute_cluster_properties(self, compute_cluster_properties: dict) -> dict: + """Validate compute cluster properties and return a normalized copy with hardware tier + IDs in the {value: str} object form expected by the API.""" + if not helpers.is_compute_cluster_properties_supported(self._version): + raise exceptions.UnsupportedFieldException( + f"'compute_cluster_properties' is not supported in Domino {self._version}." + ) + + required_keys = [ + "clusterType", + "computeEnvironmentId", + "masterHardwareTierId", + "workerHardwareTierId", + "workerCount", + ] + required_key_overrides = {("masterHardwareTierId", "MPI"): False} + + for key in required_keys: + key_required = required_key_overrides.get( + (key, compute_cluster_properties["clusterType"]), True + ) + if key_required and (key not in compute_cluster_properties): + raise exceptions.MissingRequiredFieldException( + f"{key} is required in compute_cluster_properties" + ) + + if not helpers.is_cluster_type_supported( + self._version, compute_cluster_properties["clusterType"] + ): + supported_types = [ + ct + for ct, _ in CLUSTER_TYPE_MIN_SUPPORT + if helpers.is_cluster_type_supported(self._version, ct) + ] + raise exceptions.MalformedInputException( + f"Domino {self._version} does not support cluster type {compute_cluster_properties['clusterType']}." + + f" This version of Domino supports the following cluster types: {', '.join(supported_types)}" + ) + + if "workerStorage" in compute_cluster_properties: + try: + self._validate_information_data_type( + compute_cluster_properties["workerStorage"] + ) + except Exception as e: + raise exceptions.MalformedInputException( + f"workerStorage in compute_cluster_properties failed validation: {e}" + ) + + if compute_cluster_properties["workerCount"] < 1: + raise exceptions.MalformedInputException( + "compute_cluster_properties.workerCount must be greater than 0" + ) + + if ( + "maxWorkerCount" in compute_cluster_properties + and not helpers.is_compute_cluster_autoscaling_supported(self._version) + ): + raise exceptions.UnsupportedFieldException( + f"'maxWorkerCount' is not supported in Domino {self._version}." + ) + + if "masterHardwareTierId" in compute_cluster_properties: + self._validate_hardware_tier_id( + compute_cluster_properties["masterHardwareTierId"] + ) + self._validate_hardware_tier_id( + compute_cluster_properties["workerHardwareTierId"] + ) + + normalized = compute_cluster_properties.copy() + if "masterHardwareTierId" in compute_cluster_properties: + normalized["masterHardwareTierId"] = { + "value": compute_cluster_properties["masterHardwareTierId"] + } + normalized["workerHardwareTierId"] = { + "value": compute_cluster_properties["workerHardwareTierId"] + } + return normalized + def _validate_environment_id(self, environment_id) -> bool: self.log.debug(f"Validating environment id: {environment_id}") for environment in self._useable_environments_list(): diff --git a/domino/routes.py b/domino/routes.py index 9b5c3f7b..1cce83db 100644 --- a/domino/routes.py +++ b/domino/routes.py @@ -250,6 +250,13 @@ def default_spark_setting(self, project_id): def useable_environments_list(self, project_id): return f"{self.host}/v4/projects/{project_id}/useableEnvironments" + # Scheduled Job URLs + def scheduled_jobs(self, project_id: str) -> str: + return f"{self.host}/v4/projects/{project_id}/scheduledjobs" + + def scheduled_job(self, project_id: str, scheduled_job_id: str) -> str: + return f"{self.host}/v4/projects/{project_id}/scheduledjobs/{scheduled_job_id}" + # App URLs def app_publish(self): return self._build_project_url_private_api() + "/nb/startSession" diff --git a/tests/test_scheduled_jobs.py b/tests/test_scheduled_jobs.py new file mode 100644 index 00000000..4421b7f2 --- /dev/null +++ b/tests/test_scheduled_jobs.py @@ -0,0 +1,217 @@ +""" +Tests for the scheduled jobs API. +Unit tests only — no live Domino deployment required. +""" + +import pytest + +from domino import Domino + +MOCK_PROJECT_ID = "aabbccddeeff001122334454" +MOCK_USER_ID = "aabbccddeeff001122334456" +MOCK_SCHEDULED_JOB_ID = "aabbccddeeff001122334458" + +MOCK_CRON_SCHEDULE = { + "cronString": "0 9 * * 1", + "isCustom": True, + "humanReadableCronString": "At 09:00 on Monday", +} + +MOCK_DATA_CONFIG = { + "snapshotDatasetsOnCompletion": False, + "snapshotNetAppVolumesOnCompletion": False, +} + +MOCK_SCHEDULED_JOB = { + "id": MOCK_SCHEDULED_JOB_ID, + "created": "2024-01-01T00:00:00Z", + "projectId": MOCK_PROJECT_ID, + "title": "Weekly Report", + "command": "report.py", + "schedule": MOCK_CRON_SCHEDULE, + "timezoneId": "America/New_York", + "isPaused": False, + "allowConcurrentExecution": False, + "hardwareTierIdentifier": "small-k8s", + "hardwareTierName": "Small", + "environmentRevisionSpec": "ActiveRevision", + "scheduledByUserId": MOCK_USER_ID, + "scheduledByUserName": "anyuser", + "notifyOnCompleteEmailAddresses": [], + "dataConfig": MOCK_DATA_CONFIG, +} + + +MOCK_HARDWARE_TIER_ID = "small-k8s" + +MOCK_HARDWARE_TIERS = [ + {"hardwareTier": {"id": MOCK_HARDWARE_TIER_ID, "name": "Small"}}, +] + +MOCK_ENVIRONMENT_ID = "envid123456789012345678" + +MOCK_USEABLE_ENVIRONMENTS = [ + {"id": MOCK_ENVIRONMENT_ID, "name": "Default Environment"}, +] + + +@pytest.fixture +def base_mocks(requests_mock, dummy_hostname): + requests_mock.get(f"{dummy_hostname}/version", json={"version": "9.9.9"}) + requests_mock.get( + f"{dummy_hostname}/v4/gateway/projects/findProjectByOwnerAndName" + "?ownerName=anyuser&projectName=anyproject", + json={"id": MOCK_PROJECT_ID}, + ) + requests_mock.get( + f"{dummy_hostname}/v4/users", + json=[{"id": MOCK_USER_ID, "userName": "anyuser", "email": "anyuser@example.com"}], + ) + requests_mock.get( + f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/hardwareTiers", + json=MOCK_HARDWARE_TIERS, + ) + requests_mock.get( + f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/useableEnvironments", + json={"environments": MOCK_USEABLE_ENVIRONMENTS}, + ) + yield + + +def test_scheduled_jobs_list(requests_mock, dummy_hostname, base_mocks): + requests_mock.get( + f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/scheduledjobs", + json=[MOCK_SCHEDULED_JOB], + ) + d = Domino(host=dummy_hostname, project="anyuser/anyproject", api_key="whatever") + result = d.scheduled_jobs_list() + assert isinstance(result, list) + assert len(result) == 1 + assert result[0]["id"] == MOCK_SCHEDULED_JOB_ID + + +def test_scheduled_jobs_list_with_explicit_project_id(requests_mock, dummy_hostname, base_mocks): + other_project_id = "bbccddeeff0011223344aabb" + requests_mock.get( + f"{dummy_hostname}/v4/projects/{other_project_id}/scheduledjobs", + json=[], + ) + d = Domino(host=dummy_hostname, project="anyuser/anyproject", api_key="whatever") + result = d.scheduled_jobs_list(project_id=other_project_id) + assert result == [] + + +def test_scheduled_job_get(requests_mock, dummy_hostname, base_mocks): + requests_mock.get( + f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/scheduledjobs/{MOCK_SCHEDULED_JOB_ID}", + json=MOCK_SCHEDULED_JOB, + ) + d = Domino(host=dummy_hostname, project="anyuser/anyproject", api_key="whatever") + result = d.scheduled_job_get(MOCK_SCHEDULED_JOB_ID) + assert result["id"] == MOCK_SCHEDULED_JOB_ID + assert result["title"] == "Weekly Report" + assert result["command"] == "report.py" + + +def test_scheduled_job_create(requests_mock, dummy_hostname, base_mocks): + requests_mock.post( + f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/scheduledjobs", + json=MOCK_SCHEDULED_JOB, + ) + d = Domino(host=dummy_hostname, project="anyuser/anyproject", api_key="whatever") + result = d.scheduled_job_create( + title="Weekly Report", + command="report.py", + cron_string="0 9 * * 1", + timezone_id="America/New_York", + hardware_tier_identifier="small-k8s", + ) + assert result["id"] == MOCK_SCHEDULED_JOB_ID + assert result["title"] == "Weekly Report" + + # Verify the request body contained required fields + last_request = requests_mock.last_request + body = last_request.json() + assert body["title"] == "Weekly Report" + assert body["command"] == "report.py" + assert body["schedule"]["cronString"] == "0 9 * * 1" + assert body["schedule"]["isCustom"] is True + assert body["timezoneId"] == "America/New_York" + assert body["hardwareTierIdentifier"] == "small-k8s" + assert body["environmentRevisionSpec"] == "ActiveRevision" + assert body["scheduledByUserId"] == MOCK_USER_ID + assert body["isPaused"] is False + assert body["allowConcurrentExecution"] is False + assert body["notifyOnCompleteEmailAddresses"] == [] + + +def test_scheduled_job_create_with_optional_fields(requests_mock, dummy_hostname, base_mocks): + requests_mock.post( + f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/scheduledjobs", + json=MOCK_SCHEDULED_JOB, + ) + d = Domino(host=dummy_hostname, project="anyuser/anyproject", api_key="whatever") + d.scheduled_job_create( + title="Weekly Report", + command="report.py", + cron_string="0 9 * * 1", + timezone_id="America/New_York", + hardware_tier_identifier="small-k8s", + environment_revision_spec="LatestRevision", + is_paused=True, + allow_concurrent_execution=True, + is_custom_schedule=False, + notify_on_complete_email_addresses=["user@example.com"], + capacity_type="spot", + override_environment_id="envid123456789012345678", + external_volume_mounts=["volid123456789012345678"], + snapshot_datasets_on_completion=True, + snapshot_net_app_volumes_on_completion=False, + ) + + body = requests_mock.last_request.json() + assert body["environmentRevisionSpec"] == "LatestRevision" + assert body["isPaused"] is True + assert body["allowConcurrentExecution"] is True + assert body["schedule"]["isCustom"] is False + assert body["notifyOnCompleteEmailAddresses"] == ["user@example.com"] + assert body["capacityType"] == "spot" + assert body["overrideEnvironmentId"] == "envid123456789012345678" + assert body["externalVolumeMounts"] == ["volid123456789012345678"] + assert body["snapshotDatasetsOnCompletion"] is True + assert body["snapshotNetAppVolumesOnCompletion"] is False + + +def test_scheduled_job_update(requests_mock, dummy_hostname, base_mocks): + updated = {**MOCK_SCHEDULED_JOB, "title": "Updated Report", "isPaused": True} + requests_mock.put( + f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/scheduledjobs/{MOCK_SCHEDULED_JOB_ID}", + json=updated, + ) + d = Domino(host=dummy_hostname, project="anyuser/anyproject", api_key="whatever") + result = d.scheduled_job_update( + scheduled_job_id=MOCK_SCHEDULED_JOB_ID, + title="Updated Report", + command="report.py", + cron_string="0 9 * * 1", + timezone_id="America/New_York", + hardware_tier_identifier="small-k8s", + is_paused=True, + ) + assert result["title"] == "Updated Report" + assert result["isPaused"] is True + + body = requests_mock.last_request.json() + assert body["title"] == "Updated Report" + assert body["isPaused"] is True + assert body["scheduledByUserId"] == MOCK_USER_ID + + +def test_scheduled_job_delete(requests_mock, dummy_hostname, base_mocks): + requests_mock.delete( + f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/scheduledjobs/{MOCK_SCHEDULED_JOB_ID}", + status_code=200, + ) + d = Domino(host=dummy_hostname, project="anyuser/anyproject", api_key="whatever") + response = d.scheduled_job_delete(MOCK_SCHEDULED_JOB_ID) + assert response.status_code == 200 From 42a0a1273712af20d126a935092d867ce6e39e54 Mon Sep 17 00:00:00 2001 From: Elliott Whiting Date: Wed, 1 Jul 2026 11:49:21 +0100 Subject: [PATCH 2/4] Fix example cron expression --- domino/domino.py | 6 ++++-- tests/test_scheduled_jobs.py | 10 +++++----- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/domino/domino.py b/domino/domino.py index 02d1d7d8..c2328f1d 100644 --- a/domino/domino.py +++ b/domino/domino.py @@ -822,7 +822,8 @@ def scheduled_job_create( :param command: string Command to run (e.g. "main.py arg1 arg2") :param cron_string: string - Cron expression defining the schedule (e.g. "0 9 * * 1") + Quartz cron expression (6 fields: sec min hour dom month dow), + e.g. "0 0 9 ? * MON" for every Monday at 09:00 :param timezone_id: string IANA timezone (e.g. "America/New_York", "UTC") :param hardware_tier_identifier: string @@ -931,7 +932,8 @@ def scheduled_job_update( :param command: string Command to run (e.g. "main.py arg1 arg2") :param cron_string: string - Cron expression defining the schedule (e.g. "0 9 * * 1") + Quartz cron expression (6 fields: sec min hour dom month dow), + e.g. "0 0 9 ? * MON" for every Monday at 09:00 :param timezone_id: string IANA timezone (e.g. "America/New_York", "UTC") :param hardware_tier_identifier: string diff --git a/tests/test_scheduled_jobs.py b/tests/test_scheduled_jobs.py index 4421b7f2..2bf9421c 100644 --- a/tests/test_scheduled_jobs.py +++ b/tests/test_scheduled_jobs.py @@ -12,7 +12,7 @@ MOCK_SCHEDULED_JOB_ID = "aabbccddeeff001122334458" MOCK_CRON_SCHEDULE = { - "cronString": "0 9 * * 1", + "cronString": "0 0 9 ? * MON", "isCustom": True, "humanReadableCronString": "At 09:00 on Monday", } @@ -122,7 +122,7 @@ def test_scheduled_job_create(requests_mock, dummy_hostname, base_mocks): result = d.scheduled_job_create( title="Weekly Report", command="report.py", - cron_string="0 9 * * 1", + cron_string="0 0 9 ? * MON", timezone_id="America/New_York", hardware_tier_identifier="small-k8s", ) @@ -134,7 +134,7 @@ def test_scheduled_job_create(requests_mock, dummy_hostname, base_mocks): body = last_request.json() assert body["title"] == "Weekly Report" assert body["command"] == "report.py" - assert body["schedule"]["cronString"] == "0 9 * * 1" + assert body["schedule"]["cronString"] == "0 0 9 ? * MON" assert body["schedule"]["isCustom"] is True assert body["timezoneId"] == "America/New_York" assert body["hardwareTierIdentifier"] == "small-k8s" @@ -154,7 +154,7 @@ def test_scheduled_job_create_with_optional_fields(requests_mock, dummy_hostname d.scheduled_job_create( title="Weekly Report", command="report.py", - cron_string="0 9 * * 1", + cron_string="0 0 9 ? * MON", timezone_id="America/New_York", hardware_tier_identifier="small-k8s", environment_revision_spec="LatestRevision", @@ -193,7 +193,7 @@ def test_scheduled_job_update(requests_mock, dummy_hostname, base_mocks): scheduled_job_id=MOCK_SCHEDULED_JOB_ID, title="Updated Report", command="report.py", - cron_string="0 9 * * 1", + cron_string="0 0 9 ? * MON", timezone_id="America/New_York", hardware_tier_identifier="small-k8s", is_paused=True, From 8db988abb630696ffbd707716cce850183dec1af Mon Sep 17 00:00:00 2001 From: Elliott Whiting Date: Wed, 1 Jul 2026 12:54:18 +0100 Subject: [PATCH 3/4] Lint fixes --- domino/domino.py | 16 +++++++++++----- tests/test_scheduled_jobs.py | 12 +++++++++--- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/domino/domino.py b/domino/domino.py index c2328f1d..643ecf2f 100644 --- a/domino/domino.py +++ b/domino/domino.py @@ -587,8 +587,8 @@ def validate_is_external_volume_mounts_supported(): if external_volume_mounts is not None: validate_is_external_volume_mounts_supported() if compute_cluster_properties is not None: - validated_compute_cluster_properties = self._validate_compute_cluster_properties( - compute_cluster_properties + validated_compute_cluster_properties = ( + self._validate_compute_cluster_properties(compute_cluster_properties) ) elif on_demand_spark_cluster_properties is not None: @@ -894,7 +894,9 @@ def scheduled_job_create( if snapshot_datasets_on_completion is not None: request["snapshotDatasetsOnCompletion"] = snapshot_datasets_on_completion if snapshot_net_app_volumes_on_completion is not None: - request["snapshotNetAppVolumesOnCompletion"] = snapshot_net_app_volumes_on_completion + request[ + "snapshotNetAppVolumesOnCompletion" + ] = snapshot_net_app_volumes_on_completion if main_repo_git_ref is not None: request["mainRepoGitRef"] = main_repo_git_ref url = self._routes.scheduled_jobs(self.project_id) @@ -1004,7 +1006,9 @@ def scheduled_job_update( if snapshot_datasets_on_completion is not None: request["snapshotDatasetsOnCompletion"] = snapshot_datasets_on_completion if snapshot_net_app_volumes_on_completion is not None: - request["snapshotNetAppVolumesOnCompletion"] = snapshot_net_app_volumes_on_completion + request[ + "snapshotNetAppVolumesOnCompletion" + ] = snapshot_net_app_volumes_on_completion if main_repo_git_ref is not None: request["mainRepoGitRef"] = main_repo_git_ref url = self._routes.scheduled_job(self.project_id, scheduled_job_id) @@ -2258,7 +2262,9 @@ def _useable_environments_list(self): "environments" ] - def _validate_compute_cluster_properties(self, compute_cluster_properties: dict) -> dict: + def _validate_compute_cluster_properties( + self, compute_cluster_properties: dict + ) -> dict: """Validate compute cluster properties and return a normalized copy with hardware tier IDs in the {value: str} object form expected by the API.""" if not helpers.is_compute_cluster_properties_supported(self._version): diff --git a/tests/test_scheduled_jobs.py b/tests/test_scheduled_jobs.py index 2bf9421c..6593071c 100644 --- a/tests/test_scheduled_jobs.py +++ b/tests/test_scheduled_jobs.py @@ -65,7 +65,9 @@ def base_mocks(requests_mock, dummy_hostname): ) requests_mock.get( f"{dummy_hostname}/v4/users", - json=[{"id": MOCK_USER_ID, "userName": "anyuser", "email": "anyuser@example.com"}], + json=[ + {"id": MOCK_USER_ID, "userName": "anyuser", "email": "anyuser@example.com"} + ], ) requests_mock.get( f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/hardwareTiers", @@ -90,7 +92,9 @@ def test_scheduled_jobs_list(requests_mock, dummy_hostname, base_mocks): assert result[0]["id"] == MOCK_SCHEDULED_JOB_ID -def test_scheduled_jobs_list_with_explicit_project_id(requests_mock, dummy_hostname, base_mocks): +def test_scheduled_jobs_list_with_explicit_project_id( + requests_mock, dummy_hostname, base_mocks +): other_project_id = "bbccddeeff0011223344aabb" requests_mock.get( f"{dummy_hostname}/v4/projects/{other_project_id}/scheduledjobs", @@ -145,7 +149,9 @@ def test_scheduled_job_create(requests_mock, dummy_hostname, base_mocks): assert body["notifyOnCompleteEmailAddresses"] == [] -def test_scheduled_job_create_with_optional_fields(requests_mock, dummy_hostname, base_mocks): +def test_scheduled_job_create_with_optional_fields( + requests_mock, dummy_hostname, base_mocks +): requests_mock.post( f"{dummy_hostname}/v4/projects/{MOCK_PROJECT_ID}/scheduledjobs", json=MOCK_SCHEDULED_JOB, From 8f632c4d4977c76babd3d48344379c64b06aa030 Mon Sep 17 00:00:00 2001 From: Elliott Whiting Date: Wed, 1 Jul 2026 12:58:59 +0100 Subject: [PATCH 4/4] Lint fixes --- domino/domino.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/domino/domino.py b/domino/domino.py index 643ecf2f..c641f381 100644 --- a/domino/domino.py +++ b/domino/domino.py @@ -894,9 +894,9 @@ def scheduled_job_create( if snapshot_datasets_on_completion is not None: request["snapshotDatasetsOnCompletion"] = snapshot_datasets_on_completion if snapshot_net_app_volumes_on_completion is not None: - request[ - "snapshotNetAppVolumesOnCompletion" - ] = snapshot_net_app_volumes_on_completion + request["snapshotNetAppVolumesOnCompletion"] = ( + snapshot_net_app_volumes_on_completion + ) if main_repo_git_ref is not None: request["mainRepoGitRef"] = main_repo_git_ref url = self._routes.scheduled_jobs(self.project_id) @@ -1006,9 +1006,9 @@ def scheduled_job_update( if snapshot_datasets_on_completion is not None: request["snapshotDatasetsOnCompletion"] = snapshot_datasets_on_completion if snapshot_net_app_volumes_on_completion is not None: - request[ - "snapshotNetAppVolumesOnCompletion" - ] = snapshot_net_app_volumes_on_completion + request["snapshotNetAppVolumesOnCompletion"] = ( + snapshot_net_app_volumes_on_completion + ) if main_repo_git_ref is not None: request["mainRepoGitRef"] = main_repo_git_ref url = self._routes.scheduled_job(self.project_id, scheduled_job_id)