diff --git a/.nextchanges/bundles/pydabs-job-runs.md b/.nextchanges/bundles/pydabs-job-runs.md new file mode 100644 index 00000000000..9fe4011dc1c --- /dev/null +++ b/.nextchanges/bundles/pydabs-job-runs.md @@ -0,0 +1 @@ +* Added PyDABs (Python) support for `job_runs`: `Resources.add_job_run` and the `job_run_mutator` decorator. diff --git a/acceptance/bundle/python/job-runs-support/databricks.yml b/acceptance/bundle/python/job-runs-support/databricks.yml new file mode 100644 index 00000000000..ed73fc8431f --- /dev/null +++ b/acceptance/bundle/python/job-runs-support/databricks.yml @@ -0,0 +1,10 @@ +bundle: + name: my_project + +sync: {paths: []} # don't need to copy files + +python: + resources: + - "resources:load_resources" + mutators: + - "mutators:update_job_run" diff --git a/acceptance/bundle/python/job-runs-support/mutators.py b/acceptance/bundle/python/job-runs-support/mutators.py new file mode 100644 index 00000000000..b102db4ad7b --- /dev/null +++ b/acceptance/bundle/python/job-runs-support/mutators.py @@ -0,0 +1,14 @@ +from dataclasses import replace + +from databricks.bundles.core import job_run_mutator +from databricks.bundles.job_runs import JobRun + + +@job_run_mutator +def update_job_run(job_run: JobRun) -> JobRun: + assert isinstance(job_run.job_parameters, dict) + + return replace( + job_run, + job_parameters={**job_run.job_parameters, "mutated": "true"}, + ) diff --git a/acceptance/bundle/python/job-runs-support/out.test.toml b/acceptance/bundle/python/job-runs-support/out.test.toml new file mode 100644 index 00000000000..8feac676720 --- /dev/null +++ b/acceptance/bundle/python/job-runs-support/out.test.toml @@ -0,0 +1,4 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.DMS = ["", "true"] +EnvMatrix.PYDAB_VERSION = ["current"] diff --git a/acceptance/bundle/python/job-runs-support/output.txt b/acceptance/bundle/python/job-runs-support/output.txt new file mode 100644 index 00000000000..e26d85843d9 --- /dev/null +++ b/acceptance/bundle/python/job-runs-support/output.txt @@ -0,0 +1,47 @@ + +>>> uv run [UV_ARGS] -q [CLI] bundle validate --output json +{ + "experimental": { + "python": { + "mutators": [ + "mutators:update_job_run" + ], + "resources": [ + "resources:load_resources" + ] + } + }, + "resources": { + "job_runs": { + "my_run": { + "job_id": "${resources.jobs.my_job.id}", + "job_parameters": { + "environment": "test", + "mutated": "true" + }, + "lifecycle": { + "triggers": [ + { + "on_bundle_deploy": true + } + ] + } + } + }, + "jobs": { + "my_job": { + "deployment": { + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/my_project/default/state/metadata.json" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "max_concurrent_runs": 1, + "name": "My Job", + "queue": { + "enabled": true + } + } + } + } +} diff --git a/acceptance/bundle/python/job-runs-support/resources.py b/acceptance/bundle/python/job-runs-support/resources.py new file mode 100644 index 00000000000..f82a78fc8cf --- /dev/null +++ b/acceptance/bundle/python/job-runs-support/resources.py @@ -0,0 +1,22 @@ +from databricks.bundles.core import Resources, Variable +from databricks.bundles.job_runs import JobRun, JobRunLifecycle, JobRunTrigger +from databricks.bundles.jobs import Job + + +def load_resources() -> Resources: + resources = Resources() + + resources.add_job( + "my_job", + Job(name="My Job"), + ) + resources.add_job_run( + "my_run", + JobRun( + job_id=Variable(path="resources.jobs.my_job.id", type=int), + job_parameters={"environment": "test"}, + lifecycle=JobRunLifecycle(triggers=[JobRunTrigger(on_bundle_deploy=True)]), + ), + ) + + return resources diff --git a/acceptance/bundle/python/job-runs-support/script b/acceptance/bundle/python/job-runs-support/script new file mode 100644 index 00000000000..e273fb45a53 --- /dev/null +++ b/acceptance/bundle/python/job-runs-support/script @@ -0,0 +1,5 @@ + +trace uv run $UV_ARGS -q $CLI bundle validate --output json | \ + jq "pick(.experimental.python, .resources)" + +rm -fr .databricks __pycache__ diff --git a/acceptance/bundle/python/job-runs-support/test.toml b/acceptance/bundle/python/job-runs-support/test.toml new file mode 100644 index 00000000000..56c3e19efc8 --- /dev/null +++ b/acceptance/bundle/python/job-runs-support/test.toml @@ -0,0 +1,7 @@ +Cloud = false # tests don't interact with APIs + +# job_runs are only supported in the current version of the wheel +EnvMatrix.PYDAB_VERSION = ["current"] + +# job_runs are only supported on the direct deployment engine +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] diff --git a/python/codegen/codegen/generated_wiring.py b/python/codegen/codegen/generated_wiring.py index b9ae99015d0..b1321921757 100644 --- a/python/codegen/codegen/generated_wiring.py +++ b/python/codegen/codegen/generated_wiring.py @@ -56,7 +56,7 @@ def _wired_resources() -> list[_WiredResource]: resources.append( _WiredResource( class_name=class_name, - singular_name=class_name.lower(), + singular_name=packages.get_snake_case_name(ref), plural_name=namespace, model_module=packages.get_package(namespace, ref), ) diff --git a/python/codegen/codegen/packages.py b/python/codegen/codegen/packages.py index f5c2a53fb39..25cadc70d0e 100644 --- a/python/codegen/codegen/packages.py +++ b/python/codegen/codegen/packages.py @@ -4,6 +4,7 @@ # All supported resource types and their namespace RESOURCE_NAMESPACE = { "resources.Job": "jobs", + "resources.JobRun": "job_runs", "resources.Pipeline": "pipelines", "resources.Catalog": "catalogs", "resources.Schema": "schemas", @@ -41,6 +42,11 @@ def get_class_name(ref: str) -> str: return RENAMES.get(name, name) +def get_snake_case_name(ref: str) -> str: + class_name = get_class_name(ref) + return re.sub(r"(? bool: return ref in RESOURCE_TYPES @@ -60,7 +66,6 @@ def get_package(namespace: str, ref: str) -> Optional[str]: if full_name in PRIMITIVES: return None - [_, name] = full_name.split(".") - package_name = re.sub(r"(? "tuple[_ResourceType, ...]": from databricks.bundles.core._generated import ( alerts, catalogs, + job_runs, jobs, pipelines, schemas, @@ -54,6 +61,7 @@ def _all_resource_types() -> "tuple[_ResourceType, ...]": return ( alerts._resource_type(), catalogs._resource_type(), + job_runs._resource_type(), jobs._resource_type(), pipelines._resource_type(), schemas._resource_type(), diff --git a/python/databricks/bundles/core/_generated/job_runs.py b/python/databricks/bundles/core/_generated/job_runs.py new file mode 100644 index 00000000000..243a27a931f --- /dev/null +++ b/python/databricks/bundles/core/_generated/job_runs.py @@ -0,0 +1,115 @@ +# Code generated by pydabs-codegen. DO NOT EDIT. + +from collections.abc import Callable +from typing import TYPE_CHECKING, Optional, overload + +from databricks.bundles.core._bundle import Bundle +from databricks.bundles.core._location import Location +from databricks.bundles.core._resource_mutator import ResourceMutator +from databricks.bundles.core._transform import _transform + +if TYPE_CHECKING: + from databricks.bundles.core._resource_type import _ResourceType + from databricks.bundles.job_runs._models.job_run import JobRun, JobRunParam + + +def _resource_type() -> "_ResourceType": + from databricks.bundles.core._resource_type import _ResourceType + from databricks.bundles.job_runs._models.job_run import JobRun + + return _ResourceType( + resource_type=JobRun, + singular_name="job_run", + plural_name="job_runs", + ) + + +class _JobRunResources: + """ + Generated job_run accessors, mixed into Resources. + """ + + # Provided by the Resources subclass; declared here so the generated methods + # below type-check. + _resources: dict[str, dict] + + if TYPE_CHECKING: + + def add_location(self, path: tuple[str, ...], location: Location) -> None: ... + + def add_diagnostic_error( + self, + msg: str, + *, + detail: Optional[str] = None, + path: Optional[tuple[str, ...]] = None, + location: Optional[Location] = None, + ) -> None: ... + + @property + def job_runs(self) -> dict[str, "JobRun"]: + return self._resources["job_runs"] + + def add_job_run( + self, + resource_name: str, + job_run: "JobRunParam", + *, + location: Optional[Location] = None, + ) -> None: + """ + Adds the resource job_run to the collection of resources. Resource name must be unique across all job_runs. + + :param resource_name: unique identifier for the job_run + :param job_run: the job_run to add, can be JobRun or dict + :param location: optional location of the job_run in the source code + """ + from databricks.bundles.job_runs._models.job_run import JobRun + + job_run = _transform(JobRun, job_run) + path = ("resources", "job_runs", resource_name) + location = location or Location.from_stack_frame(depth=1) + + if self._resources["job_runs"].get(resource_name): + self.add_diagnostic_error( + msg=f"Duplicate resource name '{resource_name}' for resource 'job_run'. Resource names must be unique.", + location=location, + path=path, + ) + else: + if location: + self.add_location(path, location) + + self._resources["job_runs"][resource_name] = job_run + + +@overload +def job_run_mutator( + function: Callable[[Bundle, "JobRun"], "JobRun"], +) -> ResourceMutator["JobRun"]: ... + + +@overload +def job_run_mutator( + function: Callable[["JobRun"], "JobRun"], +) -> ResourceMutator["JobRun"]: ... + + +def job_run_mutator(function: Callable) -> ResourceMutator["JobRun"]: + """ + Decorator for defining mutator for job_runs. Function should return a new instance of the job_run + with the desired changes, instead of mutating the input job_run. + + Example: + + .. code-block:: python + + @job_run_mutator + def my_job_run_mutator(bundle: Bundle, job_run: JobRun) -> JobRun: + return replace(job_run, ...) + + :param function: Function that mutates job_runs. + """ + from databricks.bundles.job_runs._models.job_run import JobRun + + return ResourceMutator(resource_type=JobRun, function=function) diff --git a/python/databricks/bundles/job_runs/__init__.py b/python/databricks/bundles/job_runs/__init__.py new file mode 100644 index 00000000000..093f2201409 --- /dev/null +++ b/python/databricks/bundles/job_runs/__init__.py @@ -0,0 +1,48 @@ +# Code generated by pydabs-codegen. DO NOT EDIT. + +__all__ = [ + "JobRun", + "JobRunDict", + "JobRunLifecycle", + "JobRunLifecycleDict", + "JobRunLifecycleParam", + "JobRunParam", + "JobRunTrigger", + "JobRunTriggerDict", + "JobRunTriggerParam", + "PerformanceTarget", + "PerformanceTargetParam", + "PipelineParams", + "PipelineParamsDict", + "PipelineParamsParam", + "QueueSettings", + "QueueSettingsDict", + "QueueSettingsParam", +] + + +from databricks.bundles.job_runs._models.job_run import JobRun, JobRunDict, JobRunParam +from databricks.bundles.job_runs._models.job_run_lifecycle import ( + JobRunLifecycle, + JobRunLifecycleDict, + JobRunLifecycleParam, +) +from databricks.bundles.job_runs._models.job_run_trigger import ( + JobRunTrigger, + JobRunTriggerDict, + JobRunTriggerParam, +) +from databricks.bundles.job_runs._models.performance_target import ( + PerformanceTarget, + PerformanceTargetParam, +) +from databricks.bundles.job_runs._models.pipeline_params import ( + PipelineParams, + PipelineParamsDict, + PipelineParamsParam, +) +from databricks.bundles.job_runs._models.queue_settings import ( + QueueSettings, + QueueSettingsDict, + QueueSettingsParam, +) diff --git a/python/databricks/bundles/job_runs/_models/job_run.py b/python/databricks/bundles/job_runs/_models/job_run.py new file mode 100644 index 00000000000..82397df768c --- /dev/null +++ b/python/databricks/bundles/job_runs/_models/job_run.py @@ -0,0 +1,136 @@ +# Code generated by pydabs-codegen. DO NOT EDIT. + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, TypedDict + +from databricks.bundles.core._resource import Resource +from databricks.bundles.core._transform import _transform +from databricks.bundles.core._transform_to_json import _transform_to_json_value +from databricks.bundles.core._variable import ( + VariableOr, + VariableOrDict, + VariableOrList, + VariableOrOptional, +) +from databricks.bundles.job_runs._models.job_run_lifecycle import ( + JobRunLifecycle, + JobRunLifecycleParam, +) +from databricks.bundles.job_runs._models.performance_target import ( + PerformanceTarget, + PerformanceTargetParam, +) +from databricks.bundles.job_runs._models.pipeline_params import ( + PipelineParams, + PipelineParamsParam, +) +from databricks.bundles.job_runs._models.queue_settings import ( + QueueSettings, + QueueSettingsParam, +) + +if TYPE_CHECKING: + from typing_extensions import Self + + +@dataclass(kw_only=True) +class JobRun(Resource): + """""" + + job_id: VariableOr[int] + """ + The ID of the job to be executed + """ + + job_parameters: VariableOrDict[str] = field(default_factory=dict) + """ + Job-level parameters used in the run. for example `"param": "overriding_val"` + """ + + lifecycle: VariableOrOptional[JobRunLifecycle] = None + """ + Settings that control the deployment lifecycle of the resource, such as preventing it from being destroyed and when the run re-fires. + """ + + only: VariableOrList[str] = field(default_factory=list) + """ + A list of task keys to run inside of the job. If this field is not provided, all tasks in the job will be run. + + Prefix a task key with `+` to also run its upstream tasks, or suffix it with `+` to also run its downstream tasks. + For example, `+my_task` runs `my_task` and everything upstream of it, `my_task+` runs `my_task` and everything + downstream of it, and `+my_task+` runs both. A task key with no `+` runs only that task. + """ + + performance_target: VariableOrOptional[PerformanceTarget] = None + """ + The performance mode on a serverless job. The performance target determines the level of compute performance or cost-efficiency for the run. This field overrides the performance target defined on the job level. + + * `STANDARD`: Enables cost-efficient execution of serverless workloads. + * `PERFORMANCE_OPTIMIZED`: Prioritizes fast startup and execution times through rapid scaling and optimized cluster performance. + """ + + pipeline_params: VariableOrOptional[PipelineParams] = None + """ + Controls whether the pipeline should perform a full refresh + """ + + queue: VariableOrOptional[QueueSettings] = None + """ + The queue settings of the run. + """ + + @classmethod + def from_dict(cls, value: "JobRunDict") -> "Self": + return _transform(cls, value) + + def as_dict(self) -> "JobRunDict": + return _transform_to_json_value(self) # type:ignore + + +class JobRunDict(TypedDict, total=False): + """""" + + job_id: VariableOr[int] + """ + The ID of the job to be executed + """ + + job_parameters: VariableOrDict[str] + """ + Job-level parameters used in the run. for example `"param": "overriding_val"` + """ + + lifecycle: VariableOrOptional[JobRunLifecycleParam] + """ + Settings that control the deployment lifecycle of the resource, such as preventing it from being destroyed and when the run re-fires. + """ + + only: VariableOrList[str] + """ + A list of task keys to run inside of the job. If this field is not provided, all tasks in the job will be run. + + Prefix a task key with `+` to also run its upstream tasks, or suffix it with `+` to also run its downstream tasks. + For example, `+my_task` runs `my_task` and everything upstream of it, `my_task+` runs `my_task` and everything + downstream of it, and `+my_task+` runs both. A task key with no `+` runs only that task. + """ + + performance_target: VariableOrOptional[PerformanceTargetParam] + """ + The performance mode on a serverless job. The performance target determines the level of compute performance or cost-efficiency for the run. This field overrides the performance target defined on the job level. + + * `STANDARD`: Enables cost-efficient execution of serverless workloads. + * `PERFORMANCE_OPTIMIZED`: Prioritizes fast startup and execution times through rapid scaling and optimized cluster performance. + """ + + pipeline_params: VariableOrOptional[PipelineParamsParam] + """ + Controls whether the pipeline should perform a full refresh + """ + + queue: VariableOrOptional[QueueSettingsParam] + """ + The queue settings of the run. + """ + + +JobRunParam = JobRunDict | JobRun diff --git a/python/databricks/bundles/job_runs/_models/job_run_lifecycle.py b/python/databricks/bundles/job_runs/_models/job_run_lifecycle.py new file mode 100644 index 00000000000..cb673b0245d --- /dev/null +++ b/python/databricks/bundles/job_runs/_models/job_run_lifecycle.py @@ -0,0 +1,54 @@ +# Code generated by pydabs-codegen. DO NOT EDIT. + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, TypedDict + +from databricks.bundles.core._transform import _transform +from databricks.bundles.core._transform_to_json import _transform_to_json_value +from databricks.bundles.core._variable import VariableOrList, VariableOrOptional +from databricks.bundles.job_runs._models.job_run_trigger import ( + JobRunTrigger, + JobRunTriggerParam, +) + +if TYPE_CHECKING: + from typing_extensions import Self + + +@dataclass(kw_only=True) +class JobRunLifecycle: + """""" + + prevent_destroy: VariableOrOptional[bool] = None + """ + Lifecycle setting to prevent the resource from being destroyed. + """ + + triggers: VariableOrList[JobRunTrigger] = field(default_factory=list) + """ + Conditions that re-fire this job run (in addition to configuration changes). + """ + + @classmethod + def from_dict(cls, value: "JobRunLifecycleDict") -> "Self": + return _transform(cls, value) + + def as_dict(self) -> "JobRunLifecycleDict": + return _transform_to_json_value(self) # type:ignore + + +class JobRunLifecycleDict(TypedDict, total=False): + """""" + + prevent_destroy: VariableOrOptional[bool] + """ + Lifecycle setting to prevent the resource from being destroyed. + """ + + triggers: VariableOrList[JobRunTriggerParam] + """ + Conditions that re-fire this job run (in addition to configuration changes). + """ + + +JobRunLifecycleParam = JobRunLifecycleDict | JobRunLifecycle diff --git a/python/databricks/bundles/job_runs/_models/job_run_trigger.py b/python/databricks/bundles/job_runs/_models/job_run_trigger.py new file mode 100644 index 00000000000..50e7c0fe12c --- /dev/null +++ b/python/databricks/bundles/job_runs/_models/job_run_trigger.py @@ -0,0 +1,40 @@ +# Code generated by pydabs-codegen. DO NOT EDIT. + +from dataclasses import dataclass +from typing import TYPE_CHECKING, TypedDict + +from databricks.bundles.core._transform import _transform +from databricks.bundles.core._transform_to_json import _transform_to_json_value +from databricks.bundles.core._variable import VariableOrOptional + +if TYPE_CHECKING: + from typing_extensions import Self + + +@dataclass(kw_only=True) +class JobRunTrigger: + """""" + + on_bundle_deploy: VariableOrOptional[bool] = None + """ + If true, re-fire the run on every bundle deploy. Incompatible with lifecycle.prevent_destroy. + """ + + @classmethod + def from_dict(cls, value: "JobRunTriggerDict") -> "Self": + return _transform(cls, value) + + def as_dict(self) -> "JobRunTriggerDict": + return _transform_to_json_value(self) # type:ignore + + +class JobRunTriggerDict(TypedDict, total=False): + """""" + + on_bundle_deploy: VariableOrOptional[bool] + """ + If true, re-fire the run on every bundle deploy. Incompatible with lifecycle.prevent_destroy. + """ + + +JobRunTriggerParam = JobRunTriggerDict | JobRunTrigger diff --git a/python/databricks/bundles/job_runs/_models/performance_target.py b/python/databricks/bundles/job_runs/_models/performance_target.py new file mode 100644 index 00000000000..8dbe7e4a435 --- /dev/null +++ b/python/databricks/bundles/job_runs/_models/performance_target.py @@ -0,0 +1,20 @@ +# Code generated by pydabs-codegen. DO NOT EDIT. + +from enum import Enum +from typing import Literal + + +class PerformanceTarget(Enum): + """ + PerformanceTarget defines how performant (lower latency) or cost efficient the execution of run on serverless compute should be. + The performance mode on the job or pipeline should map to a performance setting that is passed to Cluster Manager + (see cluster-common PerformanceTarget). + """ + + PERFORMANCE_OPTIMIZED = "PERFORMANCE_OPTIMIZED" + STANDARD = "STANDARD" + + +PerformanceTargetParam = ( + Literal["PERFORMANCE_OPTIMIZED", "STANDARD"] | PerformanceTarget +) diff --git a/python/databricks/bundles/job_runs/_models/pipeline_params.py b/python/databricks/bundles/job_runs/_models/pipeline_params.py new file mode 100644 index 00000000000..0d84734931d --- /dev/null +++ b/python/databricks/bundles/job_runs/_models/pipeline_params.py @@ -0,0 +1,82 @@ +# Code generated by pydabs-codegen. DO NOT EDIT. + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, TypedDict + +from databricks.bundles.core._transform import _transform +from databricks.bundles.core._transform_to_json import _transform_to_json_value +from databricks.bundles.core._variable import VariableOrList, VariableOrOptional + +if TYPE_CHECKING: + from typing_extensions import Self + + +@dataclass(kw_only=True) +class PipelineParams: + """""" + + full_refresh: VariableOrOptional[bool] = None + """ + If true, triggers a full refresh on the spark declarative pipeline. + """ + + full_refresh_selection: VariableOrList[str] = field(default_factory=list) + """ + [Beta] A list of tables to update with fullRefresh. + """ + + refresh_flow_selection: VariableOrList[str] = field(default_factory=list) + """ + [Beta] Flow names to selectively refresh. These are unioned with other selective refresh + options (refresh_selection, full_refresh_selection) to determine the final set of flows to refresh. + """ + + refresh_selection: VariableOrList[str] = field(default_factory=list) + """ + [Beta] A list of tables to update without fullRefresh. + """ + + reset_checkpoint_selection: VariableOrList[str] = field(default_factory=list) + """ + [Beta] A list of streaming flows to reset checkpoints without clearing data. + """ + + @classmethod + def from_dict(cls, value: "PipelineParamsDict") -> "Self": + return _transform(cls, value) + + def as_dict(self) -> "PipelineParamsDict": + return _transform_to_json_value(self) # type:ignore + + +class PipelineParamsDict(TypedDict, total=False): + """""" + + full_refresh: VariableOrOptional[bool] + """ + If true, triggers a full refresh on the spark declarative pipeline. + """ + + full_refresh_selection: VariableOrList[str] + """ + [Beta] A list of tables to update with fullRefresh. + """ + + refresh_flow_selection: VariableOrList[str] + """ + [Beta] Flow names to selectively refresh. These are unioned with other selective refresh + options (refresh_selection, full_refresh_selection) to determine the final set of flows to refresh. + """ + + refresh_selection: VariableOrList[str] + """ + [Beta] A list of tables to update without fullRefresh. + """ + + reset_checkpoint_selection: VariableOrList[str] + """ + [Beta] A list of streaming flows to reset checkpoints without clearing data. + """ + + +PipelineParamsParam = PipelineParamsDict | PipelineParams diff --git a/python/databricks/bundles/job_runs/_models/queue_settings.py b/python/databricks/bundles/job_runs/_models/queue_settings.py new file mode 100644 index 00000000000..a72921aed96 --- /dev/null +++ b/python/databricks/bundles/job_runs/_models/queue_settings.py @@ -0,0 +1,40 @@ +# Code generated by pydabs-codegen. DO NOT EDIT. + +from dataclasses import dataclass +from typing import TYPE_CHECKING, TypedDict + +from databricks.bundles.core._transform import _transform +from databricks.bundles.core._transform_to_json import _transform_to_json_value +from databricks.bundles.core._variable import VariableOr + +if TYPE_CHECKING: + from typing_extensions import Self + + +@dataclass(kw_only=True) +class QueueSettings: + """""" + + enabled: VariableOr[bool] + """ + If true, enable queueing for the job. This is a required field. + """ + + @classmethod + def from_dict(cls, value: "QueueSettingsDict") -> "Self": + return _transform(cls, value) + + def as_dict(self) -> "QueueSettingsDict": + return _transform_to_json_value(self) # type:ignore + + +class QueueSettingsDict(TypedDict, total=False): + """""" + + enabled: VariableOr[bool] + """ + If true, enable queueing for the job. This is a required field. + """ + + +QueueSettingsParam = QueueSettingsDict | QueueSettings diff --git a/python/databricks_tests/core/public_api.txt b/python/databricks_tests/core/public_api.txt index 222665796df..48189284bb0 100644 --- a/python/databricks_tests/core/public_api.txt +++ b/python/databricks_tests/core/public_api.txt @@ -16,6 +16,7 @@ __all__ = [ alert_mutator, catalog_mutator, job_mutator, + job_run_mutator, load_resources_from_current_package_module, load_resources_from_module, load_resources_from_modules, @@ -71,6 +72,7 @@ class Resources: def add_diagnostic_warning(self, msg: str, *, detail: Union[str, None] = None, path: Union[tuple[str, ...], None] = None, location: Union[Location, None] = None) -> None def add_diagnostics(self, other: Diagnostics) -> None def add_job(self, resource_name: str, job: JobParam, *, location: Union[Location, None] = None) -> None + def add_job_run(self, resource_name: str, job_run: JobRunParam, *, location: Union[Location, None] = None) -> None def add_location(self, path: tuple[str, ...], location: Location) -> None def add_pipeline(self, resource_name: str, pipeline: PipelineParam, *, location: Union[Location, None] = None) -> None def add_resource(self, resource_name: str, resource: Resource, *, location: Union[Location, None] = None) -> None @@ -80,6 +82,7 @@ class Resources: @property alerts -> dict[str, Alert] @property catalogs -> dict[str, Catalog] @property diagnostics -> Diagnostics + @property job_runs -> dict[str, JobRun] @property jobs -> dict[str, Job] @property pipelines -> dict[str, Pipeline] @property schemas -> dict[str, Schema] @@ -114,6 +117,10 @@ def catalog_mutator(function: Callable) -> ResourceMutator[Catalog] @overload def job_mutator(function: Callable[[Job], Job]) -> ResourceMutator[Job] def job_mutator(function: Callable) -> ResourceMutator[Job] +@overload def job_run_mutator(function: Callable[[Bundle, JobRun], JobRun]) -> ResourceMutator[JobRun] +@overload def job_run_mutator(function: Callable[[JobRun], JobRun]) -> ResourceMutator[JobRun] +def job_run_mutator(function: Callable) -> ResourceMutator[JobRun] + def load_resources_from_current_package_module() -> Resources def load_resources_from_module(module: module) -> Resources @@ -140,6 +147,7 @@ def volume_mutator(function: Callable) -> ResourceMutator[Volume] singular_name=alert plural_name=alerts resource_type=Alert singular_name=catalog plural_name=catalogs resource_type=Catalog singular_name=job plural_name=jobs resource_type=Job +singular_name=job_run plural_name=job_runs resource_type=JobRun singular_name=pipeline plural_name=pipelines resource_type=Pipeline singular_name=schema plural_name=schemas resource_type=Schema singular_name=volume plural_name=volumes resource_type=Volume diff --git a/python/databricks_tests/core/test_resources.py b/python/databricks_tests/core/test_resources.py index ee2ab7ec405..fe00f441d2d 100644 --- a/python/databricks_tests/core/test_resources.py +++ b/python/databricks_tests/core/test_resources.py @@ -18,6 +18,7 @@ alert_mutator, catalog_mutator, job_mutator, + job_run_mutator, pipeline_mutator, schema_mutator, volume_mutator, @@ -26,6 +27,7 @@ from databricks.bundles.core._resource import Resource from databricks.bundles.core._resource_mutator import ResourceMutator from databricks.bundles.core._resource_type import _ResourceType +from databricks.bundles.job_runs import JobRun, JobRunLifecycle, JobRunTrigger from databricks.bundles.jobs._models.job import Job from databricks.bundles.pipelines._models.pipeline import Pipeline from databricks.bundles.schemas._models.schema import Schema @@ -51,6 +53,25 @@ class TestCase: ), resource_types[Job], ), + ( + TestCase( + add_resource=Resources.add_job_run, + dict_example={ + "job_id": 123, + "job_parameters": {"environment": "test"}, + "lifecycle": {"triggers": [{"on_bundle_deploy": True}]}, + }, + dataclass_example=JobRun( + job_id=123, + job_parameters={"environment": "test"}, + lifecycle=JobRunLifecycle( + triggers=[JobRunTrigger(on_bundle_deploy=True)] + ), + ), + mutator=job_run_mutator, + ), + resource_types[JobRun], + ), ( TestCase( add_resource=Resources.add_pipeline, diff --git a/python/databricks_tests/test_build.py b/python/databricks_tests/test_build.py index 14fce619757..a4aed80e1e2 100644 --- a/python/databricks_tests/test_build.py +++ b/python/databricks_tests/test_build.py @@ -30,6 +30,7 @@ job_mutator, pipeline_mutator, ) +from databricks.bundles.job_runs import JobRun from databricks.bundles.jobs import Job from databricks.bundles.pipelines._models.pipeline import Pipeline @@ -234,6 +235,9 @@ def test_append_resources(): "job_0": {"name": "job_0"}, "job_1": {"name": "job_1"}, }, + "job_runs": { + "run_0": {"job_id": 1}, + }, "pipelines": { "pipeline_0": {"name": "pipeline_0"}, }, @@ -243,6 +247,8 @@ def test_append_resources(): resources = Resources() resources.add_job("job_1", Job(name="new name", description="new description")) resources.add_job("job_2", Job(name="job_2")) + resources.add_job_run("run_0", JobRun(job_id=2)) + resources.add_job_run("run_1", JobRun(job_id=3)) resources.add_pipeline("pipeline_1", Pipeline(name="pipeline_1")) out = _append_resources(input, resources) @@ -255,6 +261,10 @@ def test_append_resources(): "job_1": {"name": "new name", "description": "new description"}, "job_2": {"name": "job_2"}, }, + "job_runs": { + "run_0": {"job_id": 2}, + "run_1": {"job_id": 3}, + }, "pipelines": { "pipeline_0": {"name": "pipeline_0"}, "pipeline_1": {"name": "pipeline_1"}, @@ -271,6 +281,10 @@ def test_load_resources_from_input(): "job_0": {"name": "Job 0"}, "job_1": {"name": "Job 1"}, }, + "job_runs": { + "run_0": {"job_id": 1}, + "run_1": {"job_id": 2}, + }, "pipelines": { "pipeline_0": {"name": "Pipeline 0"}, "pipeline_1": {"name": "Pipeline 1"}, @@ -286,6 +300,11 @@ def test_load_resources_from_input(): "job_1": Job(name="Job 1"), } + assert resources.job_runs == { + "run_0": JobRun(job_id=1), + "run_1": JobRun(job_id=2), + } + assert resources.pipelines == { "pipeline_0": Pipeline(name="Pipeline 0"), "pipeline_1": Pipeline(name="Pipeline 1"),