dcae-cli/dcae_cli/util/__init__.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   3 # ================================================================================
   4 # Copyright (c) 2017 AT&T Intellectual Property. All rights reserved.
   5 # ================================================================================
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #      http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17 # ============LICENSE_END=========================================================
  18 #
  19 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20
  21 # -*- coding: utf-8 -*-
  22 """
  23 Provides reusable utilites
  24 """
  25 import os
  26 import json
  27 import sys
  28 import errno
  29 import contextlib
  30 import requests
  31
  32 import six
  33 import click
  34
  35 from dcae_cli.util.exc import DcaeException, FileNotFoundError
  36
  37
  38 APP_NAME = 'dcae-cli'
  39
  40
  41 def get_app_dir():
  42     '''Returns the absolute directory path for dcae cli aux files'''
  43     return click.get_app_dir(APP_NAME)
  44
  45
  46 def makedirs(path, exist_ok=True):
  47     '''Emulates Python 3.2+ os.makedirs functionality'''
  48     try:
  49         os.makedirs(path, exist_ok=exist_ok)
  50     except TypeError:
  51         try:
  52             os.makedirs(path)
  53         except OSError as e:
  54             if e.errno == errno.EEXIST and not exist_ok:
  55                 raise
  56
  57
  58 def get_pref(path, init_func=None):
  59     '''Returns a general preference dict. Uses `init_func` to create a new one if the file does not exist.'''
  60     try:
  61         with open(path) as file:
  62             pref = json.load(file)
  63     except FileNotFoundError:
  64         pref = init_func() if init_func is not None else dict()
  65         write_pref(pref, path)
  66     return pref
  67
  68
  69 def pref_exists(path):
  70     return os.path.isfile(path)
  71
  72
  73 def update_pref(path, init_func=None, **kwargs):
  74     '''Sets specified key-value pairs in a preference file and returns an updated dict'''
  75     pref = get_pref(path, init_func)
  76     pref.update(kwargs)
  77     write_pref(pref, path)
  78
  79     return pref
  80
  81
  82 def write_pref(pref, path):
  83     '''Writes a preference json file to disk'''
  84     makedirs(os.path.dirname(path), exist_ok=True)
  85     with open(path, 'w') as file:
  86         json.dump(pref, file)
  87
  88
  89 def reraise_with_msg(e, msg=None, cls=None, as_dcae=False):
  90     '''Reraises exception e with an additional message prepended'''
  91     if as_dcae:
  92         cls = DcaeException
  93     traceback = sys.exc_info()[2]
  94     cls = e.__class__ if cls is None else cls
  95     new_msg = "{:}: {:}".format(msg, e) if msg else str(e)
  96     new_e = cls(new_msg)
  97     six.reraise(cls, new_e, traceback)
  98
  99
 100 def load_json(path):
 101     '''Helper function which loads a JSON file and returns a dict'''
 102     with open(path) as file:
 103         try:
 104             return json.load(file)
 105         except ValueError:
 106             raise DcaeException("File '{}' appears to be a malformed JSON.".format(path))
 107
 108
 109 def fetch_file_from_nexus(path, transform_func=json.loads):
 110     """Fetch file from nexus
 111
 112     The default behavior is to transform the response to a json.
 113     """
 114     # TODO: Source this from app's configuration [ONAP URL TBD]
 115     server_uri = "https://make-me-valid"
 116     nexus_url = "{0}/{1}".format(server_uri, path)
 117     r = requests.get(nexus_url)
 118     r.raise_for_status()
 119     if transform_func:
 120         return transform_func(r.text)
 121     else:
 122         return r.text