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 (int | str | none) — Get the jid of the job in scheduler system</>
  • jid_file (SpecPath) — The jid file of the job</>
  • rc (int) — The return code 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 (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. </>
  • 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</>
  • log(level, msg, *args, limit, limit_indicator, logger) — Log message for the jobs</>
  • prepare(proc) — Prepare the job by given process</>
  • refresh_status() (int) — Force refresh status from file system</>
  • refresh_status_async() (int) — Async version of refresh_status for batch operations</>
method

cache()

</>

write signature to signature file

method

__repr__() → str

</>

repr of the job

method

refresh_status()

</>

Force refresh status from file system

This invalidates the cache and reads the current status. Use this in polling loops to get the latest status.

Returns (int)

The current job status

method

refresh_status_async()

</>

Async version of refresh_status for batch operations

This invalidates the cache and reads the current status asynchronously. Use this in polling loops with asyncio.gather() for parallel reads.

Returns (int)

The current job status

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

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