sdk/python/arvados/commands/_util.py

   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=10, help="""
  21 Maximum number of times to retry server requests that encounter temporary
  22 failures (e.g., server down).  Default 10.""")
  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)