seaquest.utils package¶
Submodules¶
seaquest.utils.loggus module¶
- seaquest.utils.loggus._init_logging_config(logger: Logger, file_path: str)[source]¶
set the configuration of a logging object
Set up the logging format, create the console and file handlers
Parameters¶
- logger:
Logging object
- file_path: str
Path of the file that stores the logs (will be created if not exists)
- seaquest.utils.loggus.init_logger(name: str, file_path: str = './seaquest.logs', level=10) Logger[source]¶
create an object that handles the logging
Parameters¶
- name: str
Name of the logging object
- file_path: str
Path of the file that stores the logs (will be created if not exists)
- level: default debug
Logging level
Returns¶
- logger: obj
Object that handles the logging through its inherent methods
seaquest.utils.pod module¶
- seaquest.utils.pod._delete_pod(api_instance: CoreV1Api, namespace: str, pod_name: str) None[source]¶
Delete the temporary pod used to upload files to the pvc
Parameters¶
- api_instance: kubernetes client
Kubernetes client
- namespace: str
Namespace to use for creating pvc
- pod_name: str
Name of the pod to delete
Returns¶
- None
None
- seaquest.utils.pod._get_list_of_pods(api_instance: CoreV1Api, namespace: str) list[str][source]¶
Get list of pod names in a namespace
Parameters¶
- api_instance: kubernetes client
Kubernetes client
- namespace: str
Namespace to get pods from
Returns¶
- list
List of pod names
- seaquest.utils.pod._is_running(api_instance: CoreV1Api, pod_name: str, namespace: str, timeout: int) bool[source]¶
Check if the pod is done initializing and is running
Parameters¶
- api_instance: kubernetes client
Kubernetes client
- pod_name: str
Name of the pod to check
- namespace: str
Namespace to use for creating pvc
- timeout: int
Time to wait for pod to reach running state until raising error
Returns¶
- bool
True if the pod is done initializing, False otherwise
- seaquest.utils.pod._launch_pod(api_instance: CoreV1Api, namespace: str, pod_name: str, pvc_name: str) None[source]¶
Launch a pod that will be quickly deleted after the operations are done
Parameters¶
- api_instance: kubernetes client
Kubernetes client
- namespace: str
Namespace to use for creating pod
- pod_name: str
Name of pod
- pvc_name: str
Name of pvc to bind the pod to
Returns¶
None
- seaquest.utils.pod._wait_for_running_state(api_instance: CoreV1Api, namespace: str, pod_name: str, timeout: int) bool[source]¶
Wait for the pod to be in running state, else timeout after 60 seconds
Parameters¶
- api_instance: kubernetes client
Kubernetes client
- namespace: str
Namespace to use for creating pvc
- pod_name: str
Name of the pod to check
- timeout:
Time to wait for pod to reach running state until raising error
Returns¶
- bool
True if the pod is done initializing, else raise TimeoutError
- seaquest.utils.pod.make_pod_name_unique(api_instance: CoreV1Api, namespace: str, pod_name: str) str[source]¶
Make pod name unique by appending a random number if a pod with the same name exists Parameters ———- api_instance: kubernetes client
Kubernetes client
- namespace: str
Namespace to get pods from
- pod_name: str
Desired pod name
Returns¶
- str
Unique pod name
seaquest.utils.validate module¶
- seaquest.utils.validate._parse_args(argv) dict[source]¶
parse the command line arguments
Parameters¶
- argv: list
List of command line arguments
Returns¶
- args: dict
Dictionary of parsed arguments
- seaquest.utils.validate._parse_config_file(file_path: Path) dict[source]¶
parse a yaml config file
Parameters¶
- file_path: str
Path to the config file
Returns¶
- config: dict
Dictionary of parsed config file
- seaquest.utils.validate._parse_monitor_args(argv) dict[source]¶
parse the command line arguments for the monitor
Parameters¶
- argv: list
List of command line arguments
Returns¶
- args: dict
Dictionary of parsed arguments
- seaquest.utils.validate._parse_runner_args(argv) dict[source]¶
parse the command line arguments for the runner
Parameters¶
- argv: list
List of command line arguments
Returns¶
- args: dict
Dictionary of parsed arguments
- seaquest.utils.validate._update(d: dict, u: dict)[source]¶
merge two dictionaries, keeping the values of second one in case of key collision
Parameters¶
- d: dict
First dictionary
- u: dict
Second dictionary
Returns¶
- d: dict
Merged dictionary