utils#

Utility functions for the NiceGUI frontend, including database queries for jobs, kernels, and tasks.

Functions

`archive_job_by_id`(job_id[, requesting_username])	Archive a job by marking it with a timestamp (soft delete).
`cancel_job_by_id`(job_id[, requesting_username])	Set a job's status to CANCELED.
`get_id_by_uuid`(uuid)	Get the kernel ID from its UUID.
`get_job_by_id`(job_id)	Retrieve a job by its ID.
`get_job_is_validate`(job_id)	Check if a job has status VALIDATED.
`get_job_logs_by_job_id`(job_id)	Retrieve all logs for a job in chronological order.
`get_jobs`([statuses, offset, limit, sort_by, ...])	Retrieve paginated jobs with optional filtering by status and sorting.
`get_jobs_by_op`(task_name[, user_id])	Retrieve all unique job names for a given operation/task.
`get_jobs_by_user`(user_id[, statuses, ...])	Backward-compatible wrapper: returns all jobs when user_id is empty.
`get_kernel_by_id`(kernel_id)	Retrieve a kernel by its ID.
`get_kernels_by_job_id`(job_id)	Retrieve all kernels for a specific job.
`get_kernels_by_op_and_run`(op, name[, user_id])	Retrieve kernels for a specific operation and run name.
`get_op_names`([user_id])	Retrieve all unique operation/task names, optionally filtered by user.
`get_profiler_data_reference_by_job_id`(job_id)	Retrieve reference profiler data and worker info for a job.
`get_task_by_id`(task_id)	Retrieve a task by its ID.
`get_total_jobs`([statuses])	Count total non-archived jobs, optionally filtered by status.
`get_total_jobs_by_user`(user_id[, statuses])	Count total jobs for a user, optionally filtered by status.
`normalize_runtime_stats`(runtime_stats[, ...])	Normalize the three legacy shapes of Kernel.runtime_stats into one.

kernelfoundry.gui.utils.normalize_runtime_stats(runtime_stats: dict, gpu_arch: str | None = None) → dict[source]#

Normalize the three legacy shapes of Kernel.runtime_stats into one.

The database still contains rows in each of these shapes:

{gpu: {benchmark: {mean, std, speedup, …}}} (current format)
{benchmark: {mean, std, speedup, …}} (missing the GPU level)
{mean, std, …} (flat, single benchmark/gpu)

Returns the current, 3-level shape in all cases; {} if runtime_stats is empty or unrecognized. gpu_arch (kernel.gpu_arch, possibly comma-separated) fills in the GPU level for the two older shapes; falls back to “unknown”.

kernelfoundry.gui.utils.get_jobs(statuses: list[str] | None = None, offset: int = 0, limit: int = 10, sort_by: str | None = None, descending: bool = True) → list[Job][source]#: Retrieve paginated jobs with optional filtering by status and sorting.

kernelfoundry.gui.utils.get_total_jobs(statuses: list[str] | None = None) → int[source]#: Count total non-archived jobs, optionally filtered by status.

kernelfoundry.gui.utils.get_jobs_by_user(user_id: str, statuses: list[str] | None = None, offset: int = 0, limit: int = 10, sort_by: str | None = None, descending: bool = True) → list[Job][source]#: Backward-compatible wrapper: returns all jobs when user_id is empty.

kernelfoundry.gui.utils.get_total_jobs_by_user(user_id: str, statuses: list[str] | None = None) → int[source]#: Count total jobs for a user, optionally filtered by status.

kernelfoundry.gui.utils.archive_job_by_id(job_id: int, requesting_username: str | None = None) → bool | None | str[source]#: Archive a job by marking it with a timestamp (soft delete).

kernelfoundry.gui.utils.cancel_job_by_id(job_id: int, requesting_username: str | None = None) → bool | None | str[source]#

Set a job’s status to CANCELED.

Parameters:

job_id – The ID of the job to cancel.
requesting_username – If provided, only cancel if job’s task_origin matches this username.

Returns:

True if the job was found and canceled (or already canceled), None if the job was not found, “forbidden” if requesting_username does not own the job, or False on a database error.

kernelfoundry.gui.utils.get_job_is_validate(job_id: int) → bool[source]#: Check if a job has status VALIDATED.

kernelfoundry.gui.utils.get_job_logs_by_job_id(job_id: int) → list[JobLog][source]#: Retrieve all logs for a job in chronological order.

kernelfoundry.gui.utils.get_id_by_uuid(uuid: str) → int | None[source]#: Get the kernel ID from its UUID.

kernelfoundry.gui.utils.get_kernels_by_op_and_run(op: str, name: str, user_id: str | None = None)[source]#: Retrieve kernels for a specific operation and run name.

kernelfoundry.gui.utils.get_op_names(user_id: str | None = None) → list[str][source]#: Retrieve all unique operation/task names, optionally filtered by user.

kernelfoundry.gui.utils.get_jobs_by_op(task_name: str, user_id: str | None = None) → list[str][source]#: Retrieve all unique job names for a given operation/task.

kernelfoundry.gui.utils.get_kernels_by_job_id(job_id: int)[source]#: Retrieve all kernels for a specific job.

kernelfoundry.gui.utils.get_kernel_by_id(kernel_id: int) → Kernel | None[source]#: Retrieve a kernel by its ID.

kernelfoundry.gui.utils.get_job_by_id(job_id: int) → Job | None[source]#: Retrieve a job by its ID.

kernelfoundry.gui.utils.get_task_by_id(task_id: str) → Task | None[source]#: Retrieve a task by its ID.

kernelfoundry.gui.utils.get_profiler_data_reference_by_job_id(job_id: int) → tuple[source]#: Retrieve reference profiler data and worker info for a job.