jobs

tit.server.routes.jobs

/api/jobs and /api/jobs/groups — thin HTTP wrappers over :mod:tit.jobs.manager.

Every rule that decides whether a job can run (dependencies, locks, budget) lives in :mod:tit.jobs; this module only translates HTTP into :class:~tit.jobs.manager.JobManager calls, plus the group/DAG seam into :mod:tit.jobs.plans. /api/jobs/groups covers every per-subject kind (pre, sim, flex, flex_adaptive, flex_pareto, ex, mex); the concurrency cap it carries is enforced by the scheduler, never by client-side POST timing.

InputsMissing

InputsMissing(response: JSONResponse)

Bases: Exception

:func:plan_submission's refusal: response is the route's 422 body as sent.

Source code in tit/server/routes/jobs.py

def __init__(self, response: JSONResponse) -> None:
    super().__init__("Missing inputs")
    self.response = response
    self.missing: list[dict[str, Any]] = json.loads(response.body)["missing"]

submit_group

submit_group(request: Request, body: dict[str, Any] = Body(...)) -> dict[str, Any]

Submit a per-subject job group (R3).

kind=pre expands into tit.jobs.plans.plan_preprocessing's G1-G6/report DAG; sim/flex/flex_adaptive/flex_pareto/ex/mex expand into one independent job per (subject, config) entry via tit.jobs.plans.plan_per_subject.

The whole group is created queued in this one request; tit.jobs.scheduler.evaluate() runs one job per product at a time, so its jobs run one after another. A parallel_subjects field from an older client is ignored.

subject_configs (optional, additive) carries per-subject resolved configs -- a page whose config depends on the subject (an ROI resolved against that subject's atlas, a leadfield path) sends one entry per job instead of one template. Entries are matched to subject_ids by their subject_id; a subject with no entry uses config. Whatever the caller sends, each generated config's subject_id is forced to its own subject.

Source code in tit/server/routes/jobs.py

@router.post(
    "/api/jobs/groups",
    status_code=201,
    summary="Submit one job per subject with a shared group id and a scheduler-enforced cap",
    responses={
        422: {"model": MissingInputs, "description": "missing inputs or a bad body"}
    },
)
def submit_group(request: Request, body: dict[str, Any] = Body(...)) -> dict[str, Any]:
    """Submit a per-subject job group (R3).

    ``kind=pre`` expands into ``tit.jobs.plans.plan_preprocessing``'s G1-G6/report DAG;
    ``sim``/``flex``/``flex_adaptive``/``flex_pareto``/``ex``/``mex`` expand into one independent
    job per ``(subject, config)`` entry via ``tit.jobs.plans.plan_per_subject``.

    The whole group is created queued in this one request; tit.jobs.scheduler.evaluate() runs one
    job per product at a time, so its jobs run one after another. A `parallel_subjects` field from
    an older client is ignored.

    `subject_configs` (optional, additive) carries per-subject resolved configs -- a page whose
    config depends on the subject (an ROI resolved against that subject's atlas, a leadfield
    path) sends one entry per job instead of one template. Entries are matched to
    `subject_ids` by their `subject_id`; a subject with no entry uses `config`. Whatever the
    caller sends, each generated config's `subject_id` is forced to its own subject.
    """
    from tit.jobs.plans import GROUP_KINDS

    kind = body.get("kind")
    if kind not in GROUP_KINDS:
        raise HTTPException(
            status_code=422,
            detail=f"kind must be one of {GROUP_KINDS} for a job group, got {kind!r}",
        )
    config = body.get("config")
    if not isinstance(config, dict):
        raise HTTPException(
            status_code=422, detail="config must be an object and subject_ids an array"
        )
    subject_ids = _checked_subject_ids(body.get("subject_ids"))
    created_by = _created_by(body)
    subject_configs = body.get("subject_configs")
    # An agent sends only what it chose; the rest is the run pages' (tit.server.app_defaults).
    if created_by == "agent":
        config = with_app_defaults(kind, config)
        if isinstance(subject_configs, list):
            subject_configs = [
                (
                    {**e, "config": with_app_defaults(kind, e.get("config"))}
                    if isinstance(e, dict)
                    else e
                )
                for e in subject_configs
            ]
    tags = body.get("tags") or []
    if not isinstance(tags, list):
        raise HTTPException(status_code=422, detail="tags must be an array")
    try:
        planned = plan_submission(
            _manager(request),
            kind,
            config,
            subject_ids,
            subject_configs=subject_configs,
            tags=tags,
            overwrite=bool(body.get("overwrite", False)),
        )
    except InputsMissing as exc:
        return exc.response
    return _manager(request).submit_plan(planned, created_by=created_by)

runs_once_for_all_subjects

runs_once_for_all_subjects(kind: str, config: dict[str, Any]) -> bool

True when one job of kind with config covers all its subjects (see :data:ONCE_FOR_ALL_SUBJECTS_KINDS), and for an analyzer config in group mode.

Source code in tit/server/routes/jobs.py

def runs_once_for_all_subjects(kind: str, config: dict[str, Any]) -> bool:
    """True when one job of *kind* with *config* covers all its subjects (see
    :data:`ONCE_FOR_ALL_SUBJECTS_KINDS`), and for an analyzer config in group mode."""
    return kind in ONCE_FOR_ALL_SUBJECTS_KINDS or (
        kind == "analyzer" and config.get("mode") == "group"
    )

plan_submission

plan_submission(manager: JobManager, kind: str, config: dict[str, Any], subject_ids: list[str], *, subject_configs: Any = None, tags: list[str] | None = None, overwrite: bool = False) -> list[Any]

Every check POST /api/jobs/groups makes, returning the jobs it would submit.

Shared with the proposal engine (:mod:tit.server.proposals), which queues an approved step through exactly this path. Group kinds expand as :func:submit_group documents; a config that runs once for all its subjects (:func:runs_once_for_all_subjects) is one job, as its run page submits it; any other kind becomes one job per subject with that subject's id in its config (what the agent plugin's submit_job sends to POST /api/jobs). Raises :class:HTTPException (422/409/501) or :class:InputsMissing.

Source code in tit/server/routes/jobs.py

def plan_submission(
    manager: JobManager,
    kind: str,
    config: dict[str, Any],
    subject_ids: list[str],
    *,
    subject_configs: Any = None,
    tags: list[str] | None = None,
    overwrite: bool = False,
) -> list[Any]:
    """Every check ``POST /api/jobs/groups`` makes, returning the jobs it would submit.

    Shared with the proposal engine (:mod:`tit.server.proposals`), which queues an approved
    step through exactly this path. Group kinds expand as :func:`submit_group` documents; a
    config that runs once for all its subjects (:func:`runs_once_for_all_subjects`) is one job,
    as its run page submits it; any other kind becomes one job per subject with that subject's
    id in its config (what the agent plugin's ``submit_job`` sends to ``POST /api/jobs``).
    Raises :class:`HTTPException` (422/409/501) or :class:`InputsMissing`.
    """
    from tit.jobs.plans import GROUP_KINDS
    from tit.jobs.spec import PlannedJob

    tags = list(tags or [])
    if kind == "pre":
        _check_config(kind, config)
        _check_subjects_agree(kind, config, subject_ids)
        planned = _plan_pre_group(config, subject_ids)
        _check_freesurfer_inputs(kind, config, subject_ids)
        # The whole group's flags at once: an early stage (DICOM conversion) supplies what a
        # later one (charm) needs, which a per-stage sweep could not know.
        refused = _missing_inputs(manager, [(kind, config, subject_ids)])
    else:
        if runs_once_for_all_subjects(kind, config):
            _check_config(kind, config)
            planned = [
                PlannedJob(
                    label=f"all:{kind}",
                    kind=kind,
                    config=dict(config),
                    subject_ids=list(subject_ids),
                    tags=tags,
                    overwrite=overwrite,
                )
            ]
        elif kind in GROUP_KINDS:
            planned = _plan_generic_group(
                kind,
                config,
                subject_ids,
                {"subject_configs": subject_configs},
                tags,
                overwrite,
            )
        else:
            planned = []
            for sid in subject_ids:
                entry = {**config, "subject_id": sid}
                _check_config(kind, entry)
                planned.append(
                    PlannedJob(
                        label=f"{sid}:{kind}",
                        kind=kind,
                        config=entry,
                        subject_ids=[sid],
                        tags=tags,
                        overwrite=overwrite,
                    )
                )
        refused = _missing_inputs(
            manager, [(j.kind, j.config, j.subject_ids) for j in planned]
        )
    if refused is not None:
        raise InputsMissing(refused)
    receipts = [
        (
            job,
            planned_outputs(
                job.kind, job.config, job.subject_ids, overwrite=job.overwrite
            ),
        )
        for job in planned
    ]
    # Two jobs into one folder is refused before the overwrite question is asked: no answer
    # to "replace what is on disk?" makes such a group safe.
    _refuse_shared_output_dirs(receipts)
    refuse_outputs_of_active_jobs(
        manager, [(job.kind, job.config, outputs) for job, outputs in receipts]
    )
    for job, outputs in receipts:
        refuse_unconfirmed_overwrite(job.kind, outputs, overwrite=job.overwrite)
    return planned