dpdispatcher.utils.dpcloudserver package

dpdispatcher.utils.dpcloudserver package#

Expose helpers for the legacy Bohrium cloud-service API.

class dpdispatcher.utils.dpcloudserver.Client(email: str | None = None, password: str | None = None, debug: bool = False, ticket: str | None = None, base_url: str = 'https://bohrium.dp.tech/')[source]#

Bases: object

Call legacy Bohrium APIs for authentication, jobs, logs, and files.

Methods

download_from_url(url, save_file)

Download a URL and fail explicitly after retry exhaustion.

download

get

get_job_detail

get_job_result_url

get_log

get_tasks_list

job_create

kill

post

refresh_token

upload

download(oss_file: str, save_file: str, endpoint: str, bucket_name: str) → str[source]#
download_from_url(url: str, save_file: str | PathLike) → None[source]#

Download a URL and fail explicitly after retry exhaustion.

Parameters:
urlstr

Remote URL to download.

save_filestr or os.PathLike

Local path where the response body is written.

Raises:
RequestInfoException

If no HTTP request succeeds after three attempts. The exception includes the URL and the final transport or HTTP response details.

get(url: str, header: dict[str, str] | None = None, params: dict[str, Any] | None = None, retry: int = 5) → dict[str, Any][source]#
get_job_detail(job_id: str | int) → dict[str, Any] | None[source]#
get_job_result_url(job_id: str | int) → str | None[source]#
get_log(job_id: str | int) → str[source]#
get_tasks_list(group_id: int, per_page: int = 30) → list[dict[str, Any]][source]#
job_create(job_type: str, oss_path: str, input_data: dict[str, Any], program_id: int | None = None, group_id: int | None = None) → tuple[int, int | None][source]#
kill(job_id: str | int) → dict[str, Any] | None[source]#
post(url: str, data: dict[str, Any] | None = None, header: dict[str, str] | None = None, params: dict[str, Any] | None = None, retry: int = 5) → dict[str, Any][source]#
refresh_token(retry: int = 3) → None[source]#
upload(oss_task_zip: str, zip_task_file: str, endpoint: str, bucket_name: str) → Any[source]#

Submodules#

dpdispatcher.utils.dpcloudserver.client module#

Provide a legacy HTTP client for Bohrium job and storage operations.

class dpdispatcher.utils.dpcloudserver.client.Client(email: str | None = None, password: str | None = None, debug: bool = False, ticket: str | None = None, base_url: str = 'https://bohrium.dp.tech/')[source]#

Bases: object

Call legacy Bohrium APIs for authentication, jobs, logs, and files.

Methods

download_from_url(url, save_file)

Download a URL and fail explicitly after retry exhaustion.

download

get

get_job_detail

get_job_result_url

get_log

get_tasks_list

job_create

kill

post

refresh_token

upload

download(oss_file: str, save_file: str, endpoint: str, bucket_name: str) → str[source]#
download_from_url(url: str, save_file: str | PathLike) → None[source]#

Download a URL and fail explicitly after retry exhaustion.

Parameters:
urlstr

Remote URL to download.

save_filestr or os.PathLike

Local path where the response body is written.

Raises:
RequestInfoException

If no HTTP request succeeds after three attempts. The exception includes the URL and the final transport or HTTP response details.

get(url: str, header: dict[str, str] | None = None, params: dict[str, Any] | None = None, retry: int = 5) → dict[str, Any][source]#
get_job_detail(job_id: str | int) → dict[str, Any] | None[source]#
get_job_result_url(job_id: str | int) → str | None[source]#
get_log(job_id: str | int) → str[source]#
get_tasks_list(group_id: int, per_page: int = 30) → list[dict[str, Any]][source]#
job_create(job_type: str, oss_path: str, input_data: dict[str, Any], program_id: int | None = None, group_id: int | None = None) → tuple[int, int | None][source]#
kill(job_id: str | int) → dict[str, Any] | None[source]#
post(url: str, data: dict[str, Any] | None = None, header: dict[str, str] | None = None, params: dict[str, Any] | None = None, retry: int = 5) → dict[str, Any][source]#
refresh_token(retry: int = 3) → None[source]#
upload(oss_task_zip: str, zip_task_file: str, endpoint: str, bucket_name: str) → Any[source]#
exception dpdispatcher.utils.dpcloudserver.client.RequestInfoException[source]#

Bases: Exception

Report a failed legacy Bohrium API request.

dpdispatcher.utils.dpcloudserver.config module#

Read legacy Bohrium API endpoints and object-storage settings.

dpdispatcher.utils.dpcloudserver.retcode module#

Define return codes used by the legacy Bohrium API.

class dpdispatcher.utils.dpcloudserver.retcode.RETCODE[source]#

Bases: object

Group string constants returned by the legacy Bohrium API.

DATAERR = '2002'#
DBERR = '2000'#
IOERR = '2003'#
NODATA = '2300'#
OK = '0000'#
PARAMERR = '2101'#
PWDERR = '2104'#
REQERR = '2200'#
ROLEERR = '2103'#
THIRDERR = '2001'#
TOKENINVALID = '2100'#
UNDERDEBUG = '2301'#
UNKOWNERR = '2400'#
USERERR = '2102'#
VERIFYERR = '2105'#

dpdispatcher.utils.dpcloudserver.zip_file module#

Create and extract ZIP archives for cloud file transfer.

dpdispatcher.utils.dpcloudserver.zip_file.unzip_file(zip_file: str, out_dir: str = './') → set[str][source]#

Extract a ZIP archive into a local directory.

dpdispatcher.utils.dpcloudserver.zip_file.zip_file_list(root_path: str, zip_filename: str, file_list: Iterable[str] | None = None) → str[source]#

Create a ZIP archive from paths and glob patterns below a root.

Contents