OCDocker.Workbench.Jobs module

Local subprocess launcher and tracker for Workbench API jobs (vs, pipeline, ocscore_train, ocscore_reduce). Job state is persisted under <served root>/.ocdocker-jobs/ so status survives an API restart.

Local subprocess launcher and tracker for Workbench API jobs.

exception OCDocker.Workbench.Jobs.JobError(message, *, status_code=400)[source]

Bases: Exception

HTTP-aware error raised by Workbench job execution helpers.

Parameters:

  • message (str)

  • status_code (int)

Return type:

None

__init__(message, *, status_code=400)[source]

Create a Workbench job error.

Parameters:

  • message (str) – Error message.

  • status_code (int) – HTTP status code returned by the Workbench API.

Return type:

None

class OCDocker.Workbench.Jobs.JobManager(root, *, executable='ocdocker')[source]

Bases: object

Launch and track ocdocker CLI jobs as local subprocesses.

Parameters:

  • root (str | Path)

  • executable (str)

__init__(root, *, executable='ocdocker')[source]

Bind a job manager to one served Workbench root.

Parameters:

  • root (str or pathlib.Path) – Served Workbench root. Job artifacts are written under root / ".ocdocker-jobs".

  • executable (str) – ocdocker executable used to launch jobs.

Return type:

None

plan(kind, args, *, cwd=None, manifest=None, engine='shell', cores=4, results_dir=None)[source]

Compute the command a job would run, without launching it.

Parameters:

  • kind (WorkbenchJobKind) – Job kind. Selects the ocdocker subcommand prefix, or (for "vs_campaign") the execution engine building the batch command.

  • args (Sequence[str]) – Extra command-line arguments appended after the subcommand prefix (or, for "vs_campaign", appended to every row).

  • cwd (str, pathlib.Path, or None) – Working directory the job would run in. Defaults to the served root.

  • manifest (Sequence[dict[str, Any]] or None) – Required for kind="vs_campaign": one entry per sample (sample, receptor, ligand, box, engines, optional rescoring_engines). Ignored for other kinds.

  • engine (str) – "shell" (default, a generated POSIX loop) or "snakemake" (the bundled multi-sample Snakefile, real DAG orchestration). Only meaningful for kind="vs_campaign".

  • cores (int) – --cores passed to Snakemake. Only meaningful for engine="snakemake".

  • results_dir (str or None) – Shared base output directory; each row writes to <results_dir>/<sample>. Only meaningful for kind="vs_campaign".

Returns:

JSON-safe preview: kind, command, cwd.

Return type:

dict[str, Any]

Raises:

JobError – If kind/engine is unsupported, cwd does not exist, or (for "vs_campaign") manifest is missing or malformed.

launch(kind, args, *, cwd=None, manifest=None, engine='shell', cores=4, results_dir=None)[source]

Launch one tracked job as a background subprocess.

Parameters:

  • kind (WorkbenchJobKind) – Job kind. Selects the ocdocker subcommand prefix, or (for "vs_campaign") the execution engine building the batch command.

  • args (Sequence[str]) – Extra command-line arguments appended after the subcommand prefix (or, for "vs_campaign", appended to every row).

  • cwd (str, pathlib.Path, or None) – Working directory for the launched process. Defaults to the served root.

  • manifest (Sequence[dict[str, Any]] or None) – Required for kind="vs_campaign": see plan().

  • engine (str) – "shell" or "snakemake": see plan().

  • cores (int) – --cores passed to Snakemake: see plan().

  • results_dir (str or None) – Shared base output directory: see plan().

Returns:

Tracked job record with status: "running".

Return type:

WorkbenchJobRecord

Raises:

JobError – If kind/engine is unsupported, cwd does not exist, manifest is missing/malformed for "vs_campaign", or the subprocess could not be launched.

list()[source]

Return every tracked job, most recently created first.

Returns:

Tracked job records with freshly reconciled status.

Return type:

tuple[WorkbenchJobRecord, …]

get(job_id)[source]

Return one tracked job with freshly reconciled status.

Parameters:

job_id (str) – Job identifier returned by launch().

Returns:

Tracked job record.

Return type:

WorkbenchJobRecord

Raises:

JobError – If no job with job_id is tracked.

logs(job_id, *, lines=80, max_bytes=65536)[source]

Return bounded stdout and stderr previews for one tracked job.

Parameters:

  • job_id (str) – Job identifier returned by launch().

  • lines (int) – Maximum returned lines per log file.

  • max_bytes (int) – Maximum bytes read from the end of each log file.

Returns:

Stdout preview, then stderr preview.

Return type:

tuple[RunLogFilePreview, RunLogFilePreview]

Raises:

JobError – If no job with job_id is tracked.

cancel(job_id)[source]

Terminate one running tracked job.

Parameters:

job_id (str) – Job identifier returned by launch().

Returns:

Tracked job record after cancellation. Unchanged if the job was not running.

Return type:

WorkbenchJobRecord

Raises:

JobError – If no job with job_id is tracked.

OCDocker.Workbench.Jobs.build_campaign_script(manifest, args, *, executable='ocdocker', results_dir=None)[source]

Build a POSIX shell script that runs one ocdocker command per manifest row.

Shared by JobManager.launch()/JobManager.plan() (to actually build/preview a "vs_campaign" job’s command) and OCDocker.Workbench.VSDesign.plan_vs_campaign() (to preview the same script before a job is ever created) — one source of truth for what a campaign script looks like.

Parameters:

  • manifest (Sequence[dict[str, Any]] or None) – One entry per sample: sample, row_kind ("vs" or "pipeline"), receptor, ligand, box, engines (non-empty), optional rescoring_engines.

  • args (Sequence[str]) – Flags appended to every row’s command (common to the whole batch).

  • executable (str) – ocdocker executable invoked for each row.

  • results_dir (str or None) – Every row gets its own --outdir <results_dir>/<sample> — a shared literal --outdir in args would make every row overwrite the same directory, since ocdocker vs/pipeline write straight under --outdir with no per-sample nesting of their own. Defaults to DEFAULT_CAMPAIGN_RESULTS_DIR when omitted, matching the bundled Snakefile’s default so both vs_campaign engines isolate samples the same way.

Returns:

Generated shell script. Continues past a failing row; exits non-zero only if any row failed.

Return type:

str

Raises:

JobError – If manifest is empty or a row is missing a required field.

OCDocker.Workbench.Jobs.build_campaign_snakemake_command(manifest, args, *, executable='ocdocker', cores=4, results_dir=None)[source]

Build the snakemake argv for a vs_campaign job using the bundled multi-sample Snakefile.

Shared by JobManager.launch()/JobManager.plan() (to actually build/preview a "vs_campaign", engine="snakemake" job’s command) and OCDocker.Workbench.VSDesign.plan_vs_campaign() (to preview the same command before a job is ever created). Unlike build_campaign_script(), real Snakemake DAG orchestration handles parallelism (--cores) and continuing past a failing sample (--keep-going) itself — this only builds the invocation.

Parameters:

  • manifest (Sequence[dict[str, Any]] or None) – One entry per sample: sample, row_kind ("vs" or "pipeline"), receptor, ligand, box, engines (non-empty), optional rescoring_engines.

  • args (Sequence[str]) – Flags appended to every row’s command (common to the whole batch), passed to the Snakefile as its common_args config value.

  • executable (str) – ocdocker executable invoked for each row (Snakefile ocdocker_command config value).

  • cores (int) – --cores passed to Snakemake.

  • results_dir (str or None) – Shared base output directory (Snakefile results_dir config value, default "results"); each sample’s output and .campaign_done marker land under <results_dir>/<sample>.

Returns:

snakemake argv, ready for JobManager._spawn().

Return type:

tuple[str, …]

Raises:

JobError – If manifest is empty or a row is missing a required field.