On-disk job state (TODO.md §2.3): <project>/code/ti-toolbox/jobs/<id>/{spec.json,
status.json, events.jsonl, stdout.log}.
Persisted inside the project (not a temp dir) so notebook users and a restarted server both see
the same jobs. Writes that must never leave a half-written file behind (spec.json,
status.json) go through a temp-file + :func:os.replace.
JobRegistry
JobRegistry(project_dir: str)
Filesystem operations for the job store of one project.
Source code in tit/jobs/registry.py
| def __init__(self, project_dir: str) -> None:
self.project_dir = project_dir
root = jobs_root(project_dir)
first_time = not os.path.isdir(root)
os.makedirs(root, exist_ok=True)
if first_time:
ensure_bidsignore(project_dir)
|
prune
prune(*, keep_count: int = DEFAULT_RETENTION_COUNT, keep_days: float = DEFAULT_RETENTION_DAYS, now: float | None = None) -> list[str]
Drop old, terminal jobs beyond keep_count / keep_days. Returns removed ids.
Never removes a job that is not in a terminal state (queued/running survive regardless
of age — a long-lived job should never disappear out from under it).
Source code in tit/jobs/registry.py
| def prune(
self,
*,
keep_count: int = DEFAULT_RETENTION_COUNT,
keep_days: float = DEFAULT_RETENTION_DAYS,
now: float | None = None,
) -> list[str]:
"""Drop old, terminal jobs beyond *keep_count* / *keep_days*. Returns removed ids.
Never removes a job that is not in a terminal state (queued/running survive regardless
of age — a long-lived job should never disappear out from under it).
"""
from tit.jobs.spec import (
TERMINAL_STATES,
) # local import: avoid a cycle at module load
now = now if now is not None else time.time()
cutoff = now - keep_days * 86400
records: list[tuple[str, JobStatus]] = []
for job_id in self.list_ids():
status = self.read_status(job_id)
if status is not None:
records.append((job_id, status))
terminal = [(jid, st) for jid, st in records if st.state in TERMINAL_STATES]
terminal.sort(key=lambda pair: pair[1].created_at) # oldest first
removed: list[str] = []
# Age-based
for job_id, status in terminal:
finished = status.finished_at or status.created_at
try:
# created_at/finished_at are ISO-8601; string compare works for same-format
# timestamps, but be defensive and just compare via time.time() fallback.
import datetime as _dt
ts = _dt.datetime.fromisoformat(finished).timestamp()
except ValueError:
ts = now
if ts < cutoff:
removed.append(job_id)
# Count-based: beyond keep_count oldest terminal jobs (not already marked)
remaining_terminal = [jid for jid, _ in terminal if jid not in removed]
overflow = len(remaining_terminal) - keep_count
if overflow > 0:
removed.extend(remaining_terminal[:overflow])
for job_id in removed:
self.delete(job_id)
if removed:
logger.info("job registry: pruned %d old job(s)", len(removed))
return removed
|
ensure_bidsignore
ensure_bidsignore(project_dir: str) -> None
Make sure project_dir's .bidsignore lists :data:BIDSIGNORE_LINE.
Idempotent: a no-op once the line is present. Creates .bidsignore if it doesn't exist
yet; otherwise appends to whatever is already there without touching existing lines (same
"users curate this file by hand" convention as :func:tit.pre.utils.ensure_bidsignore,
which does the same for CT files -- kept separate here rather than imported so
:mod:tit.jobs.registry never has to import :mod:tit.pre).
Source code in tit/jobs/registry.py
| def ensure_bidsignore(project_dir: str) -> None:
"""Make sure *project_dir*'s ``.bidsignore`` lists :data:`BIDSIGNORE_LINE`.
Idempotent: a no-op once the line is present. Creates ``.bidsignore`` if it doesn't exist
yet; otherwise appends to whatever is already there without touching existing lines (same
"users curate this file by hand" convention as :func:`tit.pre.utils.ensure_bidsignore`,
which does the same for CT files -- kept separate here rather than imported so
:mod:`tit.jobs.registry` never has to import :mod:`tit.pre`).
"""
target = _storage_path(project_dir, os.path.join(project_dir, ".bidsignore"))
try:
with open(target, encoding="utf-8") as fh:
existing = fh.read().splitlines()
except OSError:
existing = []
if BIDSIGNORE_LINE in existing:
return
lines = [*existing, BIDSIGNORE_LINE] if existing else [BIDSIGNORE_LINE]
with open(target, "w", encoding="utf-8") as fh:
fh.write("\n".join(lines) + "\n")
|
job_file_path
job_file_path(project_dir: str, job_id: str, filename: str) -> str
A metadata file whose existing symlinks stay within the project.
Source code in tit/jobs/registry.py
| def job_file_path(project_dir: str, job_id: str, filename: str) -> str:
"""A metadata file whose existing symlinks stay within the project."""
if not filename or filename in (".", "..") or "/" in filename or "\\" in filename:
raise PermissionError("Job metadata filename must be a single entry")
return _storage_path(
project_dir, os.path.join(job_dir(project_dir, job_id), filename)
)
|