peal.web.jobs.JobStore

class peal.web.jobs.JobStore

Bases: object

Directory-backed store of web-demo jobs, one job.json per job.

Every read and write of a job.json goes through a re-entrant lock and writes are atomic (.tmp then os.replace), so the FastAPI handlers and the Worker thread can share one store.

Parameters:

root (str) – Jobs root directory; created if missing. Job ids are 12 hex chars and each job lives in <root>/<job_id>/.

Methods

create	Create a new job directory and its initial job.json.
delete	Remove a job's whole directory (uploads, run dir, results).
estimate_wait_seconds	Seconds until the job starts (or, for a new upload, until it would start): the running job's remaining time plus one median per job ahead.
get	Load a job record.
job_dir	Directory of a job, validating the id format.
list	All readable job records under the root, oldest first.
mark_step	Update a job and record when it first reached step.
median_duration	Median wall-clock duration of the last n finished jobs.
next_queued	The oldest job with status "queued", or None.
queue	Active jobs (queued, running or awaiting feedback), oldest first.
queue_position	1 = running now, 2 = next, ...; None when not active.
stats	Queue summary shown on the landing page.
update	Merge changes into a job record and write it back atomically.

__init__(root)

job_dir(job_id)

Directory of a job, validating the id format.

Parameters:

job_id (str) – Twelve lowercase hex characters.

Returns:

<root>/<job_id>.

Return type:

str

Raises:

KeyError – If job_id is not a well-formed id (also for None).

create(fields)

Create a new job directory and its initial job.json.

Parameters:

fields (dict) – The upload form fields, stored verbatim under "fields".

Returns:

The job record with status "created", a fresh 12-hex-char id, created timestamp and None for started, finished, duration, stage and error.

Return type:

dict

get(job_id)

Load a job record.

Parameters:

job_id (str) – Job id.

Returns:

Contents of job.json.

Return type:

dict

Raises:

KeyError – If the id is malformed or the job has no job.json.

update(job_id, **changes)

Merge changes into a job record and write it back atomically.

Parameters:

• job_id (str) – Job id.

• **changes – Keys to set on the record (status, stage, error, …).

Returns:

The updated record.

Return type:

dict

list()

All readable job records under the root, oldest first.

Directories whose name is not a job id, or whose job.json is missing or corrupt, are skipped.

Returns:

Records sorted by their created timestamp.

Return type:

list of dict

delete(job_id)

Remove a job’s whole directory (uploads, run dir, results).

queue()

Active jobs (queued, running or awaiting feedback), oldest first.

Return type:

list of dict

next_queued()

The oldest job with status "queued", or None.

Return type:

dict or None

median_duration(n=10)

Median wall-clock duration of the last n finished jobs.

Parameters:

n (int) – How many of the most recent finished jobs to consider.

Returns:

Seconds; DEFAULT_JOB_SECONDS (env PEAL_WEB_DEFAULT_JOB_SECONDS, 45 minutes by default) when no job has finished yet.

Return type:

float

queue_position(job_id)

1 = running now, 2 = next, …; None when not active.

estimate_wait_seconds(job_id=None)

Seconds until the job starts (or, for a new upload, until it would start): the running job’s remaining time plus one median per job ahead. A job waiting for verdicts counts as running: it holds the GPU.

stats()

Queue summary shown on the landing page.

Returns:

queued and running counts (a job awaiting feedback counts as running), median_job_seconds and estimated_wait_seconds_for_new_job.

Return type:

dict

mark_step(job_id, step, **changes)

Update a job and record when it first reached step.

Parameters:

• job_id (str) – Job id.

• step (str) – One of STEP_ORDER.

• **changes – Further keys to set, as in update().

Returns:

The updated record.

Return type:

dict