dpdispatcher.utils package

Utils.

Subpackages

• dpdispatcher.utils.dpcloudserver package
  • Client
    • Client.download()
    • Client.download_from_url()
    • Client.get()
    • Client.get_job_detail()
    • Client.get_job_result_url()
    • Client.get_log()
    • Client.get_tasks_list()
    • Client.job_create()
    • Client.kill()
    • Client.post()
    • Client.refresh_token()
    • Client.upload()
  • Submodules
  • dpdispatcher.utils.dpcloudserver.client module
    • Client
      • Client.download()
      • Client.download_from_url()
      • Client.get()
      • Client.get_job_detail()
      • Client.get_job_result_url()
      • Client.get_log()
      • Client.get_tasks_list()
      • Client.job_create()
      • Client.kill()
      • Client.post()
      • Client.refresh_token()
      • Client.upload()
    • RequestInfoException
  • dpdispatcher.utils.dpcloudserver.config module
  • dpdispatcher.utils.dpcloudserver.retcode module
    • RETCODE
      • RETCODE.DATAERR
      • RETCODE.DBERR
      • RETCODE.IOERR
      • RETCODE.NODATA
      • RETCODE.OK
      • RETCODE.PARAMERR
      • RETCODE.PWDERR
      • RETCODE.REQERR
      • RETCODE.ROLEERR
      • RETCODE.THIRDERR
      • RETCODE.TOKENINVALID
      • RETCODE.UNDERDEBUG
      • RETCODE.UNKOWNERR
      • RETCODE.USERERR
      • RETCODE.VERIFYERR
  • dpdispatcher.utils.dpcloudserver.zip_file module
    • unzip_file()
    • zip_file_list()

Submodules

dpdispatcher.utils.archive module

Safely extract archives received from remote execution backends.

exception dpdispatcher.utils.archive.UnsafeArchiveError

Bases: ValueError

Raised when an archive member could create an unsafe filesystem entry.

dpdispatcher.utils.archive.safe_extract_tar(archive: TarFile, destination: str) → None

Safely extract regular files and directories from a tar archive.

The full manifest is validated before creating the output root. Extraction rejects links, devices, traversal, path collisions, and existing symlink components. Files are streamed to fresh temporary files and atomically replaced so existing hardlinks are not modified.

The destination must not be concurrently renamed or mutated by another process during extraction; portable Python 3.7 has no cross-platform descriptor-relative API that can close that final local TOCTOU window.

Parameters:

archivetarfile.TarFile

Open tar archive containing remote result files.

destinationstr

Directory under which every archive member must remain.

Raises:

UnsafeArchiveError

If the archive manifest or an existing output path is unsafe.

dpdispatcher.utils.archive.safe_extract_zip(archive: ZipFile, destination: str) → set[str]

Safely extract regular files and directories from a zip archive.

This applies the same manifest, symlink, and atomic-replacement guarantees as safe_extract_tar() while also validating Unix file types stored in zip metadata.

Parameters:

archivezipfile.ZipFile

Open zip archive containing remote result files.

destinationstr

Directory under which every archive member must remain.

Raises:

UnsafeArchiveError

If the archive manifest or an existing output path is unsafe.

dpdispatcher.utils.hdfs_cli module

Wrap the Hadoop CLI operations used by the HDFS context.

class dpdispatcher.utils.hdfs_cli.HDFS

Bases: object

Fundamental class for HDFS basic manipulation.

Methods

copy_from_local(local_path, to_uri)	Copy a readable local path to an HDFS URI.
copy_to_local(from_uri, local_path)	Copy one or more HDFS paths into a local directory.
exists(uri)	Return whether an HDFS URI exists.
mkdir(uri)	Create an HDFS directory and any missing parents.
move(from_uri, to_uri)	Move an HDFS path to a new URI.
read_hdfs_file(uri)	Return the decoded output of hadoop fs -text for an HDFS URI.
remove(uri)	Remove an HDFS URI recursively.

static copy_from_local(local_path: str, to_uri: str) → tuple[bool, bytes]

Copy a readable local path to an HDFS URI.

static copy_to_local(from_uri: str | list[str] | tuple[str, ...], local_path: str) → bool

Copy one or more HDFS paths into a local directory.

static exists(uri: str) → bool

Return whether an HDFS URI exists.

static mkdir(uri: str) → bool

Create an HDFS directory and any missing parents.

static move(from_uri: str, to_uri: str) → bool

Move an HDFS path to a new URI.

static read_hdfs_file(uri: str) → bytes

Return the decoded output of hadoop fs -text for an HDFS URI.

static remove(uri: str) → bool

Remove an HDFS URI recursively.

exception dpdispatcher.utils.hdfs_cli.HDFSMissingPathError

Bases: RuntimeError

Raised when Hadoop reports that a requested source path is absent.

dpdispatcher.utils.job_status module

Define scheduler-independent states for jobs and tasks.

class dpdispatcher.utils.job_status.JobStatus(value)

Bases: IntEnum

Represent normalized lifecycle states returned by machine backends.

completing = 6

failed = 7

finished = 5

running = 3

terminated = 4

unknown = 100

unsubmitted = 1

waiting = 2

dpdispatcher.utils.record module

Persist recoverable submission state in the user’s data directory.

dpdispatcher.utils.utils module

Provide hashing, authentication, retry, transfer, and script helpers.

exception dpdispatcher.utils.utils.RetrySignal

Bases: Exception

Exception to give a signal to retry the function.

dpdispatcher.utils.utils.customized_script_header_template(filename: PathLike, resources: Resources) → str

Render a user-provided scheduler header with resource values.

dpdispatcher.utils.utils.generate_totp(secret: str, period: int = 30, token_length: int = 6) → str

Generate time-based one time password (TOTP) from the secret.

Some HPCs use TOTP for two-factor authentication for safety.

Parameters:

secretstr

The encoded secret provided by the HPC. It’s usually extracted from a 2D code and base32 encoded.

periodint, default=30

Time period where the code is valid in seconds.

token_lengthint, default=6

The token length.

Returns:

token: str

The generated token.

References

lepture/otpauth

dpdispatcher.utils.utils.get_sha256(filename: str) → str

Get sha256 of a file.

Parameters:

filenamestr

The filename.

Returns:

sha256: str

The sha256.

dpdispatcher.utils.utils.hotp(key: str, period: int, token_length: int = 6, digest: str = 'sha1') → str

Generate an HMAC-based one-time password for a counter value.

dpdispatcher.utils.utils.retry(max_retry: int = 3, sleep: int | float = 60, catch_exception: type[BaseException] = <class 'dpdispatcher.utils.utils.RetrySignal'>) → Callable

Retry the function until it succeeds or fails for certain times.

Parameters:

max_retryint, default=3

The maximum retry times. If None, it will retry forever.

sleepint or float, default=60

The sleep time in seconds.

catch_exceptionException, default=Exception

The exception to catch.

Returns:

decorator: Callable

The decorator.

Examples

>>> @retry(max_retry=3, sleep=60, catch_exception=RetrySignal)
... def func():
...     raise RetrySignal("Failed")

dpdispatcher.utils.utils.rsync(from_file: str, to_file: str, port: int = 22, key_filename: str | None = None, timeout: int | float = 10, proxy_command: str | None = None) → None

Call rsync to transfer files.

Parameters:

from_filestr

SRC

to_filestr

DEST

portint, default=22

port for ssh

key_filenamestr, optional

identity file name

timeoutint, default=10

timeout for ssh

proxy_commandstr, optional

ProxyCommand to use for SSH connection

Raises:

RuntimeError

when return code is not 0

dpdispatcher.utils.utils.run_cmd_with_all_output(cmd: str | Sequence[str], shell: bool = True) → tuple[int, bytes, bytes]

Run a command and return its exit code, standard output, and error.