module

pipen.job

</>

Provide the Job class

Classes
  • Job — The job for pipen</>
class

pipen.job.Job(*args, **kwargs)

</>
Bases
xqute.job.Job pipen._job_caching.JobCaching

The job for pipen

Attributes
  • cached — Check if a job is cached</>
  • jid_file (SpecPath) — The jid file of the job</>
  • rc_file (SpecPath) — The rc file of the job</>
  • retry_dir (SpecPath) — The retry directory of the job</>
  • script_file — Get the path to script file</>
  • signature_file — Get the path to the signature file</>
  • status_file (SpecPath) — The status file of the job</>
  • stderr_file (SpecPath) — The stderr file of the job</>
  • stdout_file (SpecPath) — The stdout file of the job</>
Methods
  • __repr__() (str) — repr of the job</>
  • cache() — write signature to signature file</>
  • clean(retry) — Clean up the meta files</>
  • get_jid() (int | str | none) — Get the jid of the job in scheduler system</>
  • get_rc() (int) — The return code of the job</>
  • get_status(refresh) (int) — Query the status of the job</>
  • log(level, msg, *args, limit, limit_indicator, logger) — Log message for the jobs</>
  • prepare(proc) — Prepare the job by given process</>
  • prepare_outdir() (SpecPath) — Get the path to the output directory.</>
  • prepare_output() — Get the output data of the job</>
  • set_rc(rc) — Set the return code of the job</>
  • set_status(stat, flush) — Set the status manually</>
method

cache()

</>

write signature to signature file

method

__repr__() → str

</>

repr of the job

method

get_jid() → int | str | none

</>

Get the jid of the job in scheduler system

method

get_status(refresh=False) → int

</>

Query the status of the job

If the job is submitted, try to query it from the status file Make sure the status is updated by trap in wrapped script

Uses caching to avoid excessive file I/O. Cache is invalidated when status is explicitly set.

Parameters
  • refresh (bool, optional) — Whether to refresh the status from file
method

set_status(stat, flush=True)

</>

Set the status manually

Parameters
  • stat (int) — The status to set
  • flush (bool, optional) — Whether to flush the status to file
method

get_rc() → int

</>

The return code of the job

method

set_rc(rc)

</>

Set the return code of the job

Parameters
  • rc (int | str) — The return code
method

clean(retry=False)

</>

Clean up the meta files

Parameters
  • retry (bool, optional) — Whether clean it for retrying
method

prepare(proc)

</>

Prepare the job by given process

Primarily prepare the script, and provide cmd to the job for xqute to wrap and run

Parameters
  • proc (Proc) — the process object
method

prepare_outdir()

</>

Get the path to the output directory.

When proc.export is True, the output directory is based on the pipeline.outdir and the process name. Otherwise, it is based on the metadir.

When the job is running in a detached system (a VM, typically), this will return the mounted path to the output directory.

To access the real path, use self.outdir

Returns (SpecPath)

The path to the job output directory

method

prepare_output()

</>

Get the output data of the job

Returns

The key-value map where the keys are the output keys

method

log(level, msg, *args, limit=3, limit_indicator=True, logger=<LoggerAdapter pipen.core (WARNING)>)

</>

Log message for the jobs

Parameters
  • level (int | str) — The log level of the record
  • msg (str) — The message to log
  • *args — The arguments to format the message
  • limit (int, optional) — limitation of the log (don't log for all jobs)
  • limit_indicator (bool, optional) — Whether to show an indicator saying the loghas been limited (the level of the indicator will be DEBUG)
  • logger (LoggerAdapter, optional) — The logger used to log