kernelfoundry.eval_pipeline.database

Database initialization, utility functions and table definitions.

Functions

add_ignore_errors(item)	Adds the item to the database, ignoring any errors to not stop the program flow.
add_job(task_origin[, status])	Create and add a new Job to the database.
add_job_log(job_id, level, message[, ...])	Add a single log entry for a job.
add_job_log_batch(logs)	Add multiple log entries in a single transaction for better performance.
init(cfg)	Initialize database connections (PostgreSQL or SQLite based on DB_IP env var).
update_job_progress(job_id, progress)	Update the progress of a job in the database.
update_job_status(job_id[, status, ...])	Update the status config, and timestamps of a job in the database.

kernelfoundry.eval_pipeline.database.init(cfg)

Initialize database connections (PostgreSQL or SQLite based on DB_IP env var).

kernelfoundry.eval_pipeline.database.add_ignore_errors(item)

Adds the item to the database, ignoring any errors to not stop the program flow.

kernelfoundry.eval_pipeline.database.add_job(task_origin: str, status: str = 'INIT') → int | None

Create and add a new Job to the database.

Parameters:

status (str) – Initial status (default: “INIT”)

Returns:

The job ID if successful, None otherwise

Return type:

int | None

kernelfoundry.eval_pipeline.database.update_job_status(job_id: int, status: str | None = None, started_at: datetime | None = None, finished_at: datetime | None = None, task_id: str | None = None, config: dict | None = None, progress: float | None = None) → bool

Update the status config, and timestamps of a job in the database.

Parameters:

• job_id (int) – The ID of the job to update

• status (str) – New status (INIT, RUN, COMPLETE, FAIL)

• started_at (datetime, optional) – Timestamp when job started

• finished_at (datetime, optional) – Timestamp when job finished

• task_id (str, optional) – The task ID to associate with this job

• config (dict, optional) – Updated configuration to store in the database

• progress (float, optional) – Updated progress value between 0.0 and 1.0

Returns:

True if update was successful, False otherwise

Return type:

bool

kernelfoundry.eval_pipeline.database.update_job_progress(job_id: int, progress: float) → bool

Update the progress of a job in the database.

Parameters:

• job_id (int) – The ID of the job to update

• progress (float) – Progress value between 0.0 and 1.0

Returns:

True if update was successful, False otherwise

Return type:

bool

kernelfoundry.eval_pipeline.database.add_job_log(job_id: int, level: str, message: str, timestamp: datetime | None = None, hostname: str | None = None, kernel_uuid: str | None = None, agent_session_id: str | None = None) → int | None

Add a single log entry for a job.

Parameters:

• job_id (int) – The ID of the job this log belongs to

• level (str) – Log level (INFO, DEBUG, WARNING, ERROR)

• message (str) – The log message

• timestamp (datetime, optional) – Custom timestamp (defaults to server time)

• hostname (str | None, optional) – Hostname of the process that emitted the log entry

• kernel_uuid (str | None, optional) – UUID of the kernel associated with the log entry

• agent_session_id (str | None, optional) – Agent session identifier associated with the log entry

Returns:

The log entry ID if successful, None otherwise

Return type:

int | None

kernelfoundry.eval_pipeline.database.add_job_log_batch(logs: list[dict]) → bool

Add multiple log entries in a single transaction for better performance.

Parameters:

logs (list[dict]) – List of log dictionaries, each containing: - job_id (int): The ID of the job - hostname (str | None, optional): Hostname of the process that emitted the log entry - kernel_uuid (str | None, optional): UUID of the kernel associated with the log entry - agent_session_id (str | None, optional): Agent session identifier associated with the log entry - level (str): Log level (INFO, DEBUG, WARNING, ERROR) - message (str): The log message - timestamp (datetime, optional): Custom timestamp

Returns:

True if all logs were added successfully, False otherwise

Return type:

bool

Example:

logs = [
    {"job_id": 1, "level": "INFO", "message": "Starting task"},
    {"job_id": 1, "level": "DEBUG", "message": "Processing step 1"},
    {"job_id": 1, "level": "ERROR", "message": "Failed at step 2"},
]
add_job_log_batch(logs)

class kernelfoundry.eval_pipeline.database.Kernel(**kwargs)

id: Mapped[int]

timestamp: Mapped[datetime]

uuid: Mapped[str | None]

parent_uuid: Mapped[str | None]

task_name: Mapped[str | None]

Level_ID: Mapped[int | None]

input_code: Mapped[str | None]

input_language: Mapped[str | None]

output_code: Mapped[str | None]

output_language: Mapped[str | None]

Op_ID: Mapped[str | None]

job_name: Mapped[str | None]

prompt: Mapped[str | None]

answer: Mapped[str | None]

status: Mapped[str | None]

trial: Mapped[int | None]

runtime: Mapped[float | None]

runtime_stats: Mapped[dict[str, Any] | None]

improve_over_native: Mapped[float | None]

improve_over_compile: Mapped[float | None]

score: Mapped[int | None]

eval_log: Mapped[str | None]

version: Mapped[int | None]

index_level_0: Mapped[int | None]

language_model: Mapped[str | None]

profiler_data: Mapped[dict[str, Any] | None]

profiler_data_reference: Mapped[dict[str, Any] | None]

profiler_data_detail: Mapped[dict[str, Any] | None]

profiler_data_detail_reference: Mapped[dict[str, Any] | None]

template_results: Mapped[dict[str, Any] | None]

optimization_profile: Mapped[dict[str, Any] | None]

gpu_arch: Mapped[str | None]

eval_worker_info: Mapped[dict[str, Any] | None]

compile_worker_info: Mapped[dict[str, Any] | None]

config: Mapped[dict[str, Any] | None]

task_id: Mapped[str]

job_id: Mapped[int | None]

agent_session_id: Mapped[str | None]

db_user: Mapped[str]

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

class kernelfoundry.eval_pipeline.database.Task(**kwargs)

id: Mapped[str]

task_name: Mapped[str]

Op_ID: Mapped[str | None]

task_origin: Mapped[str]

config: Mapped[dict | None]

Level_ID: Mapped[int | None]

task_data: Mapped[dict | None]

has_build_step: Mapped[bool]

correctness_tests: Mapped[list[str] | None]

profile_tests: Mapped[list[str] | None]

evolve_block: Mapped[str | None]

reference_block: Mapped[str | None]

user_instructions_block: Mapped[str | None]

hyperparameters_buildtime: Mapped[dict | None]

hyperparameters_runtime: Mapped[dict | None]

generate_hash_id(data: str | dict) → None

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

class kernelfoundry.eval_pipeline.database.Rag(**kwargs)

id: Mapped[int]

task_name: Mapped[str | None]

language: Mapped[str]

descriptor: Mapped[str | None]

kernel_code: Mapped[str | None]

file_path: Mapped[str | None]

category: Mapped[str]

keywords: Mapped[str | None]

optimization_profile: Mapped[dict[str, float] | list[float] | None]

embedded_text: Mapped[str]

embedding: Mapped[list[float] | None]

origin: Mapped[str | None]

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

class kernelfoundry.eval_pipeline.database.Job(**kwargs)

id: Mapped[int]

task_id: Mapped[str | None]

task_origin: Mapped[str]

status: Mapped[str]

progress: Mapped[float]

created_at: Mapped[datetime]

started_at: Mapped[datetime | None]

finished_at: Mapped[datetime | None]

archived_at: Mapped[datetime | None]

config: Mapped[dict[str, Any] | None]

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

class kernelfoundry.eval_pipeline.database.JobLog(**kwargs)

id: Mapped[int]

job_id: Mapped[int]

hostname: Mapped[str | None]

kernel_uuid: Mapped[str | None]

agent_session_id: Mapped[str | None]

level: Mapped[str]

timestamp: Mapped[datetime]

message: Mapped[str]

extra: Mapped[dict[str, Any] | None]

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

Modules

tables