jobs_tools

Databricks job management utilities.

This module provides utilities for creating, scheduling, and monitoring Databricks jobs and for delivering Delta-backed Slack reminders.

Requires: - databricks-sdk - time

Author: Various contributors

jobs_tools.create_job(task_key, notebook_path, cluster_id, quartz_cron_expression, base_parameters=None, name=None)

Create a scheduled Databricks job.

Parameters:

Name	Type	Description	Default
task_key	str	Unique identifier for the task.	required
notebook_path	str	Path to the notebook to execute.	required
cluster_id	str	Databricks cluster ID to run the job on.	required
quartz_cron_expression	str	Cron expression for job schedule (Quartz format).	required
base_parameters	dict	Parameters to pass to the notebook. Defaults to {"run_date": today's date}.	None
name	str	Job name. Defaults to task_key if not provided.	None

Returns:

Name	Type	Description
int		The created job's ID.

Raises:

Type	Description
Exception	If job creation fails.

Note

• All jobs scheduled in CET timezone
• Requires WorkspaceClient credentials
• Default run_date is today's date in YYYY-MM-DD format

Example

job_id = create_job( task_key="daily_analysis", notebook_path="/Users/me/notebooks/analysis", cluster_id="0123-456789-abc", quartz_cron_expression="0 0 * * * ?" )

jobs_tools.update_job_schedule(job_id, quartz_cron_expression=None)

Update the schedule of an existing Databricks job.

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job to update.	required
quartz_cron_expression	str	New cron expression (Quartz format). If not provided, does nothing.	None

Returns:

Type	Description
	None

Raises:

Type	Description
AttributeError	If job not found.

Note

• Uses CET timezone for all schedules
• No-op if quartz_cron_expression is None

Example

update_job_schedule(123456, "0 12 * * * ?")

jobs_tools.wait_for_job_run(run_id, wait_seconds=10, maxtime=21600)

Wait for a Databricks job run to complete.

Polls the job run status at regular intervals until it reaches a terminal state (completed, failed, or skipped).

Parameters:

Name	Type	Description	Default
run_id	int	ID of the job run to wait for.	required
wait_seconds	int	Seconds to wait between status checks. Defaults to 10.	10
maxtime	int	Maximum seconds to wait before timing out. Defaults to 21600 (6 hours).	21600

Returns:

Name	Type	Description
JobState		The final state object of the completed job run.

Raises:

Type	Description
TimeoutError	If job run does not complete within maxtime seconds.

Note

Terminal states: TERMINATED, SKIPPED, INTERNAL_ERROR

Example

final_state = wait_for_job_run(987654321, wait_seconds=30) print(f"Job finished with result: {final_state.result_state}")

jobs_tools.submit_job_run(job_id, params)

Submit an ad-hoc run of an existing Databricks job.

Creates a one-time job run with the specified parameters, without affecting the job's schedule.

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job to run.	required
params	dict	Parameters to pass to the notebook. Parameter names should match those expected by the notebook.	required

Returns:

Name	Type	Description
int		The run_id of the submitted job run.

Raises:

Type	Description
AttributeError	If job not found.

Note

• Creates a one-off run with fixed name "one_off_run"
• Use wait_for_job_run(run_id) to wait for completion
• Returns the run_id for tracking purposes

Example

run_id = submit_job_run(123456, {"param1": "value1"}) final_state = wait_for_job_run(run_id)

jobs_tools.JobScheduler

Schedule and submit Databricks jobs using a Delta-backed registry.

Parameters:

Name	Type	Description	Default
schedule_table	str	Fully qualified name of the Delta table that stores scheduled jobs.	'`panel-management`.default.scheduled_jobs'

Attributes:

Name	Type	Description
schedule_table		Fully qualified schedule table name.
schedule		Pandas DataFrame containing the locally loaded schedule.
w		Databricks workspace client used to submit job runs.

Notes

This class requires an active Spark session, Delta Lake support, and valid Databricks workspace authentication. The schedule is collected to the driver as a pandas DataFrame and is intended for small scheduling tables.

jobs_tools.JobScheduler.__init__(schedule_table='`panel-management`.default.scheduled_jobs')

Initialize the scheduler and load its configured schedule table.

Parameters:

Name	Type	Description	Default
schedule_table	str	Fully qualified name of the Delta table that stores scheduled jobs.	'`panel-management`.default.scheduled_jobs'

Raises:

Type	Description
ValueError	If schedule_table is not a non-empty string.

jobs_tools.JobScheduler._validate_job_id(job_id) staticmethod

Validate and normalize a Databricks job identifier.

Parameters:

Name	Type	Description	Default
job_id	int	Databricks job identifier.	required

Returns:

Type	Description
int	Validated positive job identifier.

Raises:

Type	Description
TypeError	If job_id is not an integer.
ValueError	If job_id is not positive.

jobs_tools.JobScheduler._validate_notebook_params(notebook_params) staticmethod

Validate and copy parameters passed to a Databricks job.

Parameters:

Name	Type	Description	Default
notebook_params	Optional[Dict[str, str]]	Job parameters whose keys and values must be strings, or None.	required

Returns:

Type	Description
dict or None	A shallow copy of the validated parameters, or None.

Raises:

Type	Description
TypeError	If the value is not a dictionary or contains a non-string key or value.

jobs_tools.JobScheduler._empty_schedule() classmethod

Return an empty pandas schedule with the required columns.

Returns:

Type	Description
DataFrame	Empty frame whose columns are :attr:_SCHEDULE_COLUMNS, suitable as a starting point when no schedule table exists yet.

jobs_tools.JobScheduler._validate_schedule(schedule)

Validate the columns and keys of an in-memory schedule.

Parameters:

Name	Type	Description	Default
schedule	DataFrame	Schedule rows to validate.	required

Raises:

Type	Description
TypeError	If schedule is not a pandas DataFrame or contains invalid job parameters.
ValueError	If required columns, identifiers, statuses, or dates are invalid.

jobs_tools.JobScheduler._new_schedule_id()

Return a positive 63-bit identifier not present in memory.

Returns:

Type	Description
int	Identifier in the range 1 to 2**63 - 1 that does not collide with a schedule_id already in the in-memory schedule. Uniqueness is only checked against loaded rows, not against the Delta table.

jobs_tools.JobScheduler._write_rows(rows)

Create the schedule table or merge selected rows into it.

Parameters:

Name	Type	Description	Default
rows	DataFrame	Complete schedule rows to create or merge by schedule_id.	required

jobs_tools.JobScheduler._idempotency_token(schedule_id)

Return a stable Databricks idempotency token for a schedule row.

Parameters:

Name	Type	Description	Default
schedule_id	int	Identifier of the schedule row the token represents.	required

Returns:

Type	Description
str	SHA-256 hex digest of the table name and schedule_id. The value is deterministic, so a retried submission reuses the same token and Databricks does not start a duplicate run.

jobs_tools.JobScheduler.schedule_job(job_id, run_date, notebook_params=None, write_schedule=True)

Add a pending Databricks job run to the schedule.

Parameters:

Name	Type	Description	Default
job_id	int	Positive identifier of the Databricks job to run.	required
run_date	Union[str, datetime, date]	Date on which the job should be submitted. Strings must use YYYY-MM-DD format.	required
notebook_params	Optional[Dict[str, str]]	Optional job parameters with string keys and string values.	None
write_schedule	bool	If True, persist the new schedule row immediately.	True

Raises:

Type	Description
TypeError	If job_id or notebook_params has an invalid type.
ValueError	If job_id, run_date, or notebook_params is invalid.

jobs_tools.JobScheduler.read_schedule()

Load and validate the configured Delta schedule in memory.

Notes

The complete table is collected to the driver as a pandas DataFrame. If the table does not exist, an empty schedule with the expected columns is initialized. Calling this method discards unpersisted in-memory changes.

jobs_tools.JobScheduler.write_schedule()

Merge the complete in-memory schedule into the Delta table.

Notes

A missing table is created using the explicit schedule schema. Existing rows are matched by schedule_id and updated; unmatched rows are inserted. Normal scheduling and execution paths write only changed rows.

jobs_tools.JobScheduler.execute_scheduled_jobs(include_previous_days=False)

Submit pending jobs whose run dates are eligible.

Parameters:

Name	Type	Description	Default
include_previous_days	bool	If True, submit pending jobs dated today or earlier. Otherwise, submit only jobs dated today.	False

Notes

Each successful submission is persisted immediately with status submitted. Stable idempotency tokens prevent the same schedule row from creating another run when a submission is retried.

jobs_tools.JobScheduler.execute_job(job_id, notebook_params=None, idempotency_token=None)

Submit a configured Databricks job for immediate execution.

Parameters:

Name	Type	Description	Default
job_id	int	Positive identifier of the Databricks job to submit.	required
notebook_params	Optional[Dict[str, str]]	Optional job parameters with string keys and string values.	None
idempotency_token	Optional[str]	Optional token of at most 64 characters that prevents duplicate runs when the same request is retried.	None

Returns:

Type	Description
int	Identifier of the submitted Databricks run.

Raises:

Type	Description
TypeError	If an input has an invalid type.
ValueError	If an identifier or idempotency token is invalid.

jobs_tools.JobScheduler.get_pending_jobs(include_previous_days=False)

Return pending jobs whose run dates are eligible.

Parameters:

Name	Type	Description	Default
include_previous_days	bool	If True, include jobs dated today or earlier. Otherwise, include only jobs dated today.	False

Returns:

Type	Description
DataFrame	Eligible rows with status pending. The original schedule indices are preserved.

Raises:

Type	Description
ValueError	If the in-memory schedule contains an invalid or null run date.

jobs_tools.JobScheduler.delete_schedule(auto_confirm=False)

Delete every row from the configured schedule after confirmation.

Parameters:

Name	Type	Description	Default
auto_confirm	bool	If True, skip the interactive prompt and delete immediately.	False

Notes

Confirmed deletion removes all active Delta rows and resets the in-memory schedule. Delta transaction history may still permit recovery according to the table's retention configuration.

jobs_tools.ReminderScheduler

Store and deliver Slack reminders from a Delta table.

Parameters:

Name	Type	Description	Default
reminders_table	str	Fully qualified Delta table used to persist reminders.	'`panel-management`.default.reminders_schedule'
secret_scope	str	Databricks secret scope containing the webhook mapping.	'dame'
secret_key	str	Secret key containing a JSON object of webhook IDs and URLs.	'webhooks'
max_attempts	int	Maximum number of delivery attempts for one reminder.	3
claim_timeout_minutes	int	Minutes after which an interrupted delivery claim can be retried.	30

Notes

Reminder dates use the Europe/Copenhagen calendar. Delivery is at-least-once: row claiming prevents concurrent and repeated delivery after success, but a process failure after Slack accepts a message and before the Delta status update can still result in a retry.

jobs_tools.ReminderScheduler.__init__(reminders_table='`panel-management`.default.reminders_schedule', secret_scope='dame', secret_key='webhooks', max_attempts=3, claim_timeout_minutes=30)

Initialize the reminder registry and create it when missing.

Parameters:

Name	Type	Description	Default
reminders_table	str	Fully qualified Delta table used to persist reminders.	'`panel-management`.default.reminders_schedule'
secret_scope	str	Databricks secret scope containing the webhook mapping.	'dame'
secret_key	str	Secret key containing a JSON object of webhook IDs and URLs.	'webhooks'
max_attempts	int	Maximum number of delivery attempts for one reminder.	3
claim_timeout_minutes	int	Minutes after which an interrupted delivery claim can be retried.	30

Raises:

Type	Description
TypeError	If a configuration value has an invalid type.
ValueError	If a configuration value is empty or not positive, or an existing table has an incompatible schema.

jobs_tools.ReminderScheduler._validate_non_empty_string(value, name) staticmethod

Validate and strip a required string value.

Parameters:

Name	Type	Description	Default
value	str	Value to validate.	required
name	str	Human-readable field name used in error messages.	required

Returns:

Type	Description
str	value with surrounding whitespace removed.

Raises:

Type	Description
TypeError	If value is not a string.
ValueError	If value is empty or whitespace only.

jobs_tools.ReminderScheduler._ensure_table()

Create the reminders table or validate its existing schema.

Raises:

Type	Description
ValueError	If the existing table has missing, unexpected, or type-incompatible columns relative to :attr:_REMINDER_SCHEMA, or if it already contains a row whose status is null or outside :attr:_VALID_STATUSES.

Notes

The Delta schema and delivery statuses are part of this class's public operational contract, so an incompatible table is rejected rather than migrated.

jobs_tools.ReminderScheduler._load_webhooks()

Load and validate the webhook mapping from Databricks secrets.

Returns:

Type	Description
dict	Mapping of webhook ID to webhook URL, read from the configured secret scope and key.

Raises:

Type	Description
ValueError	If the secret does not contain a JSON object, or if any webhook ID or URL is empty.
TypeError	If any webhook ID or URL is not a string.

Notes

Webhook URLs are secret values. Callers must keep them out of logs, exceptions, and the reminders table.

jobs_tools.ReminderScheduler._new_reminder_id()

Return a positive 63-bit identifier absent from the Delta table.

Returns:

Type	Description
int	Identifier in the range 1 to 2**63 - 1 with no matching reminder_id in the reminders table.

Notes

Each candidate costs one query against the reminders table. The check is not transactional, so concurrent writers can still pick the same identifier.

jobs_tools.ReminderScheduler._reset_stale_claims()

Release delivery claims left behind by interrupted senders.

jobs_tools.ReminderScheduler.submit_reminder(reminder_date, user, message, webhook_id=_DEFAULT_WEBHOOK_ID)

Persist a Slack reminder for delivery on or after a date.

Parameters:

Name	Type	Description	Default
reminder_date	Union[datetime, date]	Copenhagen calendar date on which the reminder becomes eligible. A datetime is reduced to its calendar date.	required
user	str	Username or recipient identifier included in the webhook payload.	required
message	str	Reminder text included in the webhook payload.	required
webhook_id	str	Key used to select a URL from the configured webhook secret.	_DEFAULT_WEBHOOK_ID

Returns:

Type	Description
int	Unique reminder identifier stored in Delta.

Raises:

Type	Description
TypeError	If an argument has an invalid type.
ValueError	If a string is empty or webhook_id is not configured.

Examples:

>>> reminders = ReminderScheduler()
>>> reminders.submit_reminder(
...     date(2026, 4, 15),
...     "john.doe",
...     "Review Q2 reports",
... )

jobs_tools.ReminderScheduler.send_reminders()

Send all eligible reminders and persist each delivery outcome.

Returns:

Type	Description
dict	Counts named eligible, sent, failed, and skipped for this invocation.

Notes

Pending and previously failed reminders dated today or earlier are eligible until max_attempts is reached. Each row is claimed atomically before delivery, and successful rows are never selected again. Failures are recorded and do not stop delivery of the remaining reminders.