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.

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"))
    tags = body.get("tags") or []
    if not isinstance(tags, list):
        raise HTTPException(status_code=422, detail="tags must be an array")
    overwrite = bool(body.get("overwrite", False))

    if kind == "pre":
        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(request), [(kind, config, subject_ids)])
    else:
        planned = _plan_generic_group(kind, config, subject_ids, body, tags, overwrite)
        refused = _missing_inputs(
            _manager(request), [(j.kind, j.config, j.subject_ids) for j in planned]
        )
    if refused is not None:
        return refused
    for job in planned:
        check_overwrite_permission(
            job.kind, job.config, job.subject_ids, overwrite=job.overwrite
        )
    return _manager(request).submit_plan(planned, created_by="gui")