dpdispatcher.contexts package

dpdispatcher.contexts package#

Contexts.

Submodules#

dpdispatcher.contexts.dp_cloud_server_context module#

Stage files for the legacy Bohrium cloud-service backend.

class dpdispatcher.contexts.dp_cloud_server_context.BohriumContext(*args: Any, **kwargs: Any)[source]#

Bases: BaseContext

Transfer submissions through Bohrium object storage and cloud APIs.

Methods

`bind_submission`(submission)	Bind a submission and its derived working paths to this context.
`block_call`(cmd)	Run command with arguments.
`block_checkcall`(cmd[, asynchronously])	Run command with arguments.
`check_file_exists`(fname)	Return whether a file exists in the active execution root.
`check_finish`(proc)	Return whether an asynchronous process has finished.
`clean`()	Remove the submission-specific execution directory.
`download`(submission[, check_exists, ...])	Download completed job archives and record their extracted paths.
`load_from_dict`(context_dict)	Create a registered context from a machine configuration mapping.
`machine_arginfo`()	Generate the machine arginfo.
`machine_subfields`()	Generate the machine subfields.
`read_file`(fname)	Read text from a file relative to the execution root.
`upload`(submission)	Upload all files required by a submission to the execution root.
`write_file`(fname, write_str)	Write text to a file relative to the execution root.
`write_local_file`(fname, write_str)	Write a backend-local staging file when the context supports it.

check_home_file_exits
read_home_file
upload_job
write_home_file

alias: tuple[str, ...] = ('DpCloudServerContext', 'LebesgueContext')#

bind_submission(submission: Submission) → None[source]#: Bind a submission and its derived working paths to this context.

block_call(cmd: str) → NoReturn[source]#

Run command with arguments. Wait for command to complete.

Parameters:

cmdstr: The command to run.

Returns:

exit_status: exit code
stdin: standard inout
stdout: standard output
stderr: standard error

check_file_exists(fname: str) → bool[source]#: Return whether a file exists in the active execution root.

check_home_file_exits(fname: str) → bool[source]#

clean() → bool[source]#: Remove the submission-specific execution directory.

download(submission: Submission, check_exists: bool = False, mark_failure: bool = True, back_error: bool = False) → bool[source]#: Download completed job archives and record their extracted paths.

downloads_by_job: ClassVar[bool] = True#

classmethod load_from_dict(context_dict: dict[str, Any]) → BohriumContext[source]#: Create a registered context from a machine configuration mapping.

classmethod machine_subfields() → list[Argument][source]#

Generate the machine subfields.

Returns:

list[Argument]: machine subfields

read_file(fname: str) → str[source]#: Read text from a file relative to the execution root.

read_home_file(fname: str) → str[source]#

supports_task_completion_tags: ClassVar[bool] = False#

upload(submission: Submission) → None[source]#: Upload all files required by a submission to the execution root.

upload_job(job: Job, common_files: list[str] | None = None) → None[source]#

write_file(fname: str, write_str: str) → bool[source]#: Write text to a file relative to the execution root.

write_home_file(fname: str, write_str: str) → bool[source]#

write_local_file(fname: str, write_str: str) → str[source]#: Write a backend-local staging file when the context supports it.

dpdispatcher.contexts.dp_cloud_server_context.DpCloudServerContext#: alias of BohriumContext

dpdispatcher.contexts.dp_cloud_server_context.LebesgueContext#: alias of BohriumContext

dpdispatcher.contexts.hdfs_context module#

Stage DistributedShell submission data through HDFS archives.

class dpdispatcher.contexts.hdfs_context.HDFSContext(*args: Any, **kwargs: Any)[source]#

Bases: BaseContext

Transfer submission inputs and outputs between local storage and HDFS.

Methods

`bind_submission`(submission)	Bind a submission and its derived working paths to this context.
`block_call`(cmd)	Run command with arguments.
`block_checkcall`(cmd[, asynchronously])	Run command with arguments.
`check_file_exists`(fname)	Check whether the given file exists, often used in checking whether the belonging job has finished.
`check_finish`(proc)	Return whether an asynchronous process has finished.
`clean`()	Remove the submission-specific execution directory.
`download`(submission[, check_exists, ...])	Download backward files from HDFS root dir.
`load_from_dict`(context_dict)	Create a registered context from a machine configuration mapping.
`machine_arginfo`()	Generate the machine arginfo.
`machine_subfields`()	Generate the machine subfields.
`migrate_recovery_root`(old_remote_root, ...)	Move a recovered submission tree through the HDFS API.
`read_file`(fname)	Read text from a file relative to the execution root.
`rollback_recovery_root`(current_remote_root, ...)	Reverse a recovery migration, tolerating an already-finished rollback.
`upload`(submission[, dereference])	Upload forward files and forward command files to HDFS root dir.
`write_file`(fname, write_str)	Write text to a file relative to the execution root.
`write_local_file`(fname, write_str)	Write a backend-local staging file when the context supports it.

get_job_root

bind_submission(submission: Submission) → None[source]#: Bind a submission and its derived working paths to this context.

block_call(cmd: str) → NoReturn[source]#

Run command with arguments. Wait for command to complete.

Parameters:

cmdstr: The command to run.

Returns:

exit_status: exit code
stdin: standard inout
stdout: standard output
stderr: standard error

check_file_exists(fname: str) → bool[source]#

Check whether the given file exists, often used in checking whether the belonging job has finished.

Parameters:

fnamestring: file name to be checked

Returns:

status: boolean

clean() → None[source]#: Remove the submission-specific execution directory.

download(submission: Submission, check_exists: bool = False, mark_failure: bool = True, back_error: bool = False) → None[source]#

Download backward files from HDFS root dir.

Parameters:

submissionSubmission class instance: represents a collection of tasks, such as backward file names
check_existsbool: whether to check if the file exists
mark_failurebool: whether to mark the task as failed if the file does not exist
back_errorbool: whether to download error files

Returns:

none

downloads_by_job: ClassVar[bool] = True#

get_job_root() → str[source]#

classmethod load_from_dict(context_dict: dict[str, Any]) → HDFSContext[source]#: Create a registered context from a machine configuration mapping.

migrate_recovery_root(old_remote_root: str, new_remote_root: str, *, force: bool = False) → bool[source]#

Move a recovered submission tree through the HDFS API.

Resource-only resumes must preserve completion tags stored under the previous hash. Local filesystem operations cannot inspect or rename an HDFS URI, so use backend existence checks and an atomic HDFS move. Existing new-hash state wins to keep repeated recovery idempotent.

Parameters:

forcebool, default=False: Treat an existing destination with no source as an already-completed operation and report success. Recovery passes this only while reversing a move, so a concurrent rollback is considered successful instead of being mistaken for a failed rollback.

Returns:

bool: Whether this call moved (or, for a forced rollback, confirmed) an existing source root. Recovery uses the signal to roll the move back if subsequent rebinding fails.

read_file(fname: str) → bytes[source]#: Read text from a file relative to the execution root.

rollback_recovery_root(current_remote_root: str, previous_remote_root: str) → bool[source]#

Reverse a recovery migration, tolerating an already-finished rollback.

This dedicated hook keeps the force detail private to HDFS while allowing older third-party contexts that implement only the original two-argument migration hook to remain usable by Submission.

supports_partial_job_download: ClassVar[bool] = False#

upload(submission: Submission, dereference: bool = True) → None[source]#

Upload forward files and forward command files to HDFS root dir.

Parameters:

submissionSubmission class instance: represents a collection of tasks, such as forward file names
dereferencebool: whether to dereference symbolic links

Returns:

none

write_file(fname: str, write_str: str) → None[source]#: Write text to a file relative to the execution root.

dpdispatcher.contexts.lazy_local_context module#

Execute jobs directly inside their existing local working directory.

class dpdispatcher.contexts.lazy_local_context.LazyLocalContext(*args: Any, **kwargs: Any)[source]#

Bases: BaseContext

Run jobs in the local server and local directory.

Parameters:

local_rootstr: The local directory to store the jobs.
remote_rootstr, optional: The argument takes no effect.
remote_profiledict, optional: The remote profile. The default is {}.
*args: The arguments.
**kwargs: The keyword arguments.

Methods

`bind_submission`(submission)	Bind a submission and its derived working paths to this context.
`block_call`(cmd)	Run command with arguments.
`block_checkcall`(cmd[, asynchronously])	Run command with arguments.
`check_file_exists`(fname)	Return whether a file exists in the active execution root.
`check_finish`(proc)	Return whether an asynchronous process has finished.
`clean`()	Remove the submission-specific execution directory.
`download`(submission[, check_exists, ...])	Download declared result files from the execution root.
`load_from_dict`(context_dict)	Create a registered context from a machine configuration mapping.
`machine_arginfo`()	Generate the machine arginfo.
`machine_subfields`()	Generate the machine subfields.
`migrate_recovery_root`(old_remote_root, ...)	Keep the shared work-base directory in place during recovery.
`read_file`(fname)	Read text from a file relative to the execution root.
`upload`(submission[, dereference])	Upload all files required by a submission to the execution root.
`write_file`(fname, write_str)	Write text to a file relative to the execution root.
`write_local_file`(fname, write_str)	Write a backend-local staging file when the context supports it.

call
get_job_root
get_return

bind_submission(submission: Submission) → None[source]#: Bind a submission and its derived working paths to this context.

block_call(cmd: str) → tuple[int, None, SPRetObj, SPRetObj][source]#

Run command with arguments. Wait for command to complete.

Parameters:

cmdstr: The command to run.

Returns:

exit_status: exit code
stdin: standard inout
stdout: standard output
stderr: standard error

call(cmd: str) → Popen[source]#

check_file_exists(fname: str) → bool[source]#: Return whether a file exists in the active execution root.

check_finish(proc: Popen) → bool[source]#: Return whether an asynchronous process has finished.

clean() → None[source]#: Remove the submission-specific execution directory.

download(submission: Submission, check_exists: bool = False, mark_failure: bool = True, back_error: bool = False) → None[source]#: Download declared result files from the execution root.

get_job_root() → str[source]#

get_return(proc: Popen) → tuple[int | None, SPRetObj | None, SPRetObj | None][source]#

classmethod load_from_dict(context_dict: dict[str, Any]) → LazyLocalContext[source]#: Create a registered context from a machine configuration mapping.

migrate_recovery_root(old_remote_root: str, new_remote_root: str, *, force: bool = False) → bool[source]#

Keep the shared work-base directory in place during recovery.

LazyLocalContext deliberately reuses one directory for every submission hash, so moving it would make the subsequent bind point at a path that this backend never uses.

Returns:

bool: Always False because the shared work directory is not moved.

read_file(fname: str) → str[source]#: Read text from a file relative to the execution root.

upload(submission: Submission, dereference: bool = True) → None[source]#: Upload all files required by a submission to the execution root.

write_file(fname: str, write_str: str) → None[source]#: Write text to a file relative to the execution root.

class dpdispatcher.contexts.lazy_local_context.SPRetObj(ret: bytes)[source]#

Bases: object

Adapt subprocess byte output to the stream interface used by contexts.

Methods

read
readlines

read() → bytes[source]#

readlines() → list[str][source]#

dpdispatcher.contexts.local_context module#

Execute jobs locally while staging them in a separate directory.

class dpdispatcher.contexts.local_context.LocalContext(*args: Any, **kwargs: Any)[source]#

Bases: BaseContext

Run jobs in the local server and remote directory.

Parameters:

local_rootstr: The local directory to store the jobs.
remote_rootstr: The remote directory to store the jobs.
remote_profiledict, optional: The remote profile. The default is {}.
*args: The arguments.
**kwargs: The keyword arguments.

Methods

`bind_submission`(submission)	Bind a submission and its derived working paths to this context.
`block_call`(cmd)	Run command with arguments.
`block_checkcall`(cmd[, asynchronously])	Run command with arguments.
`check_file_exists`(fname)	Return whether a file exists in the active execution root.
`check_finish`(proc)	Return whether an asynchronous process has finished.
`clean`()	Remove the submission-specific execution directory.
`download`(submission[, check_exists, ...])	Download requested artifacts through a manifest and one copier.
`load_from_dict`(context_dict)	Create a registered context from a machine configuration mapping.
`machine_arginfo`()	Generate the machine arginfo.
`machine_subfields`()	Generate the machine subfields.
`read_file`(fname)	Read text from a file relative to the execution root.
`upload`(submission)	Stage all forward files through one validated transfer manifest.
`write_file`(fname, write_str)	Write text to a file relative to the execution root.
`write_local_file`(fname, write_str)	Write a backend-local staging file when the context supports it.

call
get_job_root
get_return

bind_submission(submission: Submission) → None[source]#: Bind a submission and its derived working paths to this context.

block_call(cmd: str) → tuple[int, None, SPRetObj, SPRetObj][source]#

Run command with arguments. Wait for command to complete.

Parameters:

cmdstr: The command to run.

Returns:

exit_status: exit code
stdin: standard inout
stdout: standard output
stderr: standard error

call(cmd: str) → Popen[source]#

check_file_exists(fname: str) → bool[source]#: Return whether a file exists in the active execution root.

check_finish(proc: Popen) → bool[source]#: Return whether an asynchronous process has finished.

clean() → None[source]#: Remove the submission-specific execution directory.

download(submission: Submission, check_exists: bool = False, mark_failure: bool = True, back_error: bool = False) → None[source]#: Download requested artifacts through a manifest and one copier.

get_job_root() → str[source]#

get_return(proc: Popen) → tuple[int | None, SPRetObj | None, SPRetObj | None][source]#

classmethod load_from_dict(context_dict: dict[str, Any]) → LocalContext[source]#: Create a registered context from a machine configuration mapping.

classmethod machine_subfields() → list[Argument][source]#

Generate the machine subfields.

Returns:

list[Argument]: machine subfields

read_file(fname: str) → str[source]#: Read text from a file relative to the execution root.

upload(submission: Submission) → None[source]#: Stage all forward files through one validated transfer manifest.

write_file(fname: str, write_str: str) → None[source]#: Write text to a file relative to the execution root.

class dpdispatcher.contexts.local_context.SPRetObj(ret: bytes)[source]#

Bases: object

Adapt subprocess byte output to the stream interface used by contexts.

Methods

read
readlines

read() → bytes[source]#

readlines() → list[str][source]#

dpdispatcher.contexts.openapi_context module#

Stage submissions for Bohrium through its SDK-based OpenAPI.

class dpdispatcher.contexts.openapi_context.OpenAPIContext(*args: Any, **kwargs: Any)[source]#

Bases: BaseContext

Transfer files with Bohrium object storage authenticated by access key.

Methods

`bind_submission`(submission)	Bind a submission and its derived working paths to this context.
`block_call`(cmd)	Run command with arguments.
`block_checkcall`(cmd[, asynchronously])	Run command with arguments.
`check_file_exists`(fname)	Return whether a file exists in the active execution root.
`check_finish`(proc)	Return whether an asynchronous process has finished.
`clean`()	Remove the submission-specific execution directory.
`download`(submission[, check_exists, ...])	Download completed Bohrium archives and record extracted paths.
`load_from_dict`(context_dict)	Create a registered context from a machine configuration mapping.
`machine_arginfo`()	Generate the machine arginfo.
`machine_subfields`()	Generate the machine subfields.
`read_file`(fname)	Read text from a file relative to the execution root.
`upload`(submission)	Upload all files required by a submission to the execution root.
`write_file`(fname, write_str)	Write text to a file relative to the execution root.
`write_local_file`(fname, write_str)	Write a backend-local staging file when the context supports it.

check_home_file_exits
read_home_file
upload_job
write_home_file

bind_submission(submission: Submission) → None[source]#: Bind a submission and its derived working paths to this context.

block_call(cmd: str) → NoReturn[source]#

Run command with arguments. Wait for command to complete.

Parameters:

cmdstr: The command to run.

Returns:

exit_status: exit code
stdin: standard inout
stdout: standard output
stderr: standard error

check_file_exists(fname: str) → bool[source]#: Return whether a file exists in the active execution root.

check_home_file_exits(fname: str) → bool[source]#

clean() → bool[source]#: Remove the submission-specific execution directory.

download(submission: Submission, check_exists: bool = False, mark_failure: bool = True, back_error: bool = False) → bool[source]#: Download completed Bohrium archives and record extracted paths.

downloads_by_job: ClassVar[bool] = True#

classmethod load_from_dict(context_dict: dict[str, Any]) → OpenAPIContext[source]#: Create a registered context from a machine configuration mapping.

read_file(fname: str) → str[source]#: Read text from a file relative to the execution root.

read_home_file(fname: str) → str[source]#

supports_task_completion_tags: ClassVar[bool] = False#

upload(submission: Submission) → None[source]#: Upload all files required by a submission to the execution root.

upload_job(job: DPJob, common_files: list[str] | None = None) → None[source]#

write_file(fname: str, write_str: str) → bool[source]#: Write text to a file relative to the execution root.

write_home_file(fname: str, write_str: str) → bool[source]#

write_local_file(fname: str, write_str: str) → str[source]#: Write a backend-local staging file when the context supports it.

dpdispatcher.contexts.ssh_context module#

Provide SSH command execution and rsync/SFTP file staging.

class dpdispatcher.contexts.ssh_context.SSHContext(*args: Any, **kwargs: Any)[source]#

Bases: BaseContext

Run submissions on a remote host reached through an SSH session.

Attributes:

sftp
ssh

Methods

`bind_submission`(submission)	Bind a submission and its derived working paths to this context.
`block_call`(cmd)	Run command with arguments.
`block_checkcall`(cmd[, asynchronously])	Run command with arguments.
`check_file_exists`(fname)	Return whether a file exists in the active execution root.
`check_finish`(proc)	Return whether an asynchronous process has finished.
`clean`()	Remove the submission-specific execution directory.
`download`(submission[, check_exists, ...])	Download selected remote files using one SFTP-backed manifest.
`load_from_dict`(context_dict)	Create a registered context from a machine configuration mapping.
`machine_arginfo`()	Generate the machine arginfo.
`machine_subfields`()	Generate the machine subfields.
`read_file`(fname)	Read text from a file relative to the execution root.
`upload`(submission[, dereference])	Upload all files required by a submission to the execution root.
`write_file`(fname, write_str)	Write text to a file relative to the execution root.
`write_local_file`(fname, write_str)	Write a backend-local staging file when the context supports it.

call
close
get_job_root
get_return
list_remote_dir

bind_submission(submission: Submission) → None[source]#: Bind a submission and its derived working paths to this context.

block_call(cmd: str) → tuple[int, Any, Any, Any][source]#

Run command with arguments. Wait for command to complete.

Parameters:

cmdstr: The command to run.

Returns:

exit_status: exit code
stdin: standard inout
stdout: standard output
stderr: standard error

call(cmd: str) → dict[str, Any][source]#

check_file_exists(fname: str) → bool[source]#: Return whether a file exists in the active execution root.

check_finish(proc: dict[str, Any]) → bool[source]#: Return whether an asynchronous process has finished.

clean() → None[source]#: Remove the submission-specific execution directory.

close() → None[source]#

download(submission: Submission, check_exists: bool = False, mark_failure: bool = True, back_error: bool = False) → None[source]#

Download selected remote files using one SFTP-backed manifest.

Remote wildcard expansion is performed against a single recursive SFTP index; the controller’s local glob implementation must never be applied to paths that only exist on the remote host.

get_job_root() → str[source]#

get_return(cmd_pipes: dict[str, Any]) → tuple[int | None, Any, Any][source]#

list_remote_dir(sftp: SFTPClient, remote_dir: str, ref_remote_root: str, result_list: list[str]) → None[source]#

classmethod load_from_dict(context_dict: dict[str, Any]) → SSHContext[source]#: Create a registered context from a machine configuration mapping.

classmethod machine_subfields() → list[Argument][source]#

Generate the machine subfields.

Returns:

list[Argument]: machine subfields

read_file(fname: str) → str[source]#: Read text from a file relative to the execution root.

property sftp: SFTPClient#

property ssh: SSHClient#

upload(submission: Submission, dereference: bool = True) → None[source]#: Upload all files required by a submission to the execution root.

write_file(fname: str, write_str: str) → None[source]#: Write text to a file relative to the execution root.

class dpdispatcher.contexts.ssh_context.SSHSession(hostname: str, username: str, password: str | None = None, port: int = 22, key_filename: str | None = None, passphrase: str | None = None, timeout: int = 10, totp_secret: str | None = None, tar_compress: bool = True, look_for_keys: bool = True, execute_command: str | None = None, proxy_command: str | None = None, archive_chunk_size: int = 0)[source]#

Bases: object

Manage a resilient Paramiko SSH connection and file-transfer helpers.

Attributes:

remote
rsync_available
sftp: Returns sftp.

Methods

inter_handler(title, instructions, prompt_list)

inter_handler: the callback for paramiko.transport.auth_interactive.

arginfo
close
ensure_alive
exec_command
get
get_ssh_client
put

static arginfo() → Argument[source]#

close() → None[source]#

ensure_alive(max_check: int = 10, sleep_time: int = 10) → None[source]#

exec_command(**kwargs: Any) → Any#

get(from_f: str, to_f: str) → SFTPAttributes | None[source]#

get_ssh_client() → SSHClient[source]#

inter_handler(title: str, instructions: str, prompt_list: list[tuple[str, bool]]) → list[str][source]#

inter_handler: the callback for paramiko.transport.auth_interactive.

The prototype for this function is defined by Paramiko, so all of the arguments need to be there, even though we don’t use ‘title’ or ‘instructions’.

The function is expected to return a tuple of data containing the responses to the provided prompts. Experimental results suggests that there will be one call of this function per prompt, but the mechanism allows for multiple prompts to be sent at once, so it’s best to assume that that can happen.

Since tuples can’t really be built on the fly, the responses are collected in a list which is then converted to a tuple when it’s time to return a value.

Experiments suggest that the username prompt never happens. This makes sense, but the Username prompt is included here just in case.

put(from_f: str, to_f: str) → SFTPAttributes | None[source]#

property remote: str#

property rsync_available: bool#

property sftp: SFTPClient#: Returns sftp. Open a new one if not existing.

dpdispatcher.contexts package

Contents

dpdispatcher.contexts package#

Submodules#

dpdispatcher.contexts.dp_cloud_server_context module#

dpdispatcher.contexts.hdfs_context module#

dpdispatcher.contexts.lazy_local_context module#

dpdispatcher.contexts.local_context module#

dpdispatcher.contexts.openapi_context module#

dpdispatcher.contexts.ssh_context module#