Source Code for Module arvados.commands._util

 1  # Copyright (C) The Arvados Authors. All rights reserved. 
 2  # 
 3  # SPDX-License-Identifier: Apache-2.0 
 4   
 5  import argparse 
 6  import errno 
 7  import os 
 8  import logging 
 9  import signal 
10  from future.utils import listitems, listvalues 
11  import sys 
12   
13 -def _pos_int(s): 
14      num = int(s) 
15      if num < 0: 
16          raise ValueError("can't accept negative value: %s" % (num,)) 
17      return num 
18   
19  retry_opt = argparse.ArgumentParser(add_help=False) 
20  retry_opt.add_argument('--retries', type=_pos_int, default=3, help=""" 
21  Maximum number of times to retry server requests that encounter temporary 
22  failures (e.g., server down).  Default 3.""") 
23   
24 -def _ignore_error(error): 
25      return None 
26   
27 -def _raise_error(error): 
28      raise error 
29   
30 -def make_home_conf_dir(path, mode=None, errors='ignore'): 
31      # Make the directory path under the user's home directory, making parent 
32      # directories as needed. 
33      # If the directory is newly created, and a mode is specified, chmod it 
34      # with those permissions. 
35      # If there's an error, return None if errors is 'ignore', else raise an 
36      # exception. 
37      error_handler = _ignore_error if (errors == 'ignore') else _raise_error 
38      tilde_path = os.path.join('~', path) 
39      abs_path = os.path.expanduser(tilde_path) 
40      if abs_path == tilde_path: 
41          return error_handler(ValueError("no home directory available")) 
42      try: 
43          os.makedirs(abs_path) 
44      except OSError as error: 
45          if error.errno != errno.EEXIST: 
46              return error_handler(error) 
47      else: 
48          if mode is not None: 
49              os.chmod(abs_path, mode) 
50      return abs_path 
51   
52  CAUGHT_SIGNALS = [signal.SIGINT, signal.SIGQUIT, signal.SIGTERM] 
53   
54 -def exit_signal_handler(sigcode, frame): 
55      logging.getLogger('arvados').error("Caught signal {}, exiting.".format(sigcode)) 
56      sys.exit(-sigcode) 
57   
58 -def install_signal_handlers(): 
59      global orig_signal_handlers 
60      orig_signal_handlers = {sigcode: signal.signal(sigcode, exit_signal_handler) 
61                              for sigcode in CAUGHT_SIGNALS} 
62   
63 -def restore_signal_handlers(): 
64      for sigcode, orig_handler in listitems(orig_signal_handlers): 
65          signal.signal(sigcode, orig_handler) 
66