ody_migr_utils Namespace Reference

Functions
def	get_ith_batch (_list, batch_index, batch_size)
def	encode_base64 (_data)
def	decode_base64 (_encode)
def	get_valid_json (_obj, **kwargs)
def	sanitize_insert_values (_values)
def	sanitize_dict (records_dict)
def	sanitize_insert_collection (_collection)
def	generate_mdx_hash (_cu, _key)
def	generate_hash (_action, _cu, _secret_key)
def	format_error_permission (_e, _file__or_script)
def	file_exc_decorator (original_file_handler)
def	get_strip_value_dict (_k, _dict, **kwargs)
def	log_progress (_progress_dict)
def	custom_css_parser (_content)

Variables
	LOGGER = logging.getLogger(__name__)

Detailed Description

This module includes utility functions.

Function Documentation

custom_css_parser()

def ody_migr_utils.custom_css_parser

(

_content

)

Simple css parser to convert css attributes to python dict.

Args:
    _content (str): main css content

Returns:
    cssdict (dict): parsed python dictionary

Alternative advanced css parsing using tinycss
**implementation starts**
import tinycss
parser = tinycss.make_parser()
stylesheet = parser.parse_stylesheet(_content)
cssdict = {}
for rule in stylesheet.rules:
    this_rule = rule.selector.as_css()
    cssdict[this_rule] = {}
    print (this_rule)
    for css_attr in rule.declarations:
        # print (css_attr)
        cssdict[this_rule][css_attr.name] = css_attr.value.as_css()
        print (css_attr.name, css_attr.value.as_css())
print(stylesheet.errors)
return cssdict
**implementation ends**

Definition at line 250 of file ody_migr_utils.py.

def custom_css_parser(_content):
    """Simple css parser to convert css attributes to python dict.

    Args:
        _content (str): main css content

    Returns:
        cssdict (dict): parsed python dictionary

    Alternative advanced css parsing using tinycss
    **implementation starts**
    import tinycss
    parser = tinycss.make_parser()
    stylesheet = parser.parse_stylesheet(_content)
    cssdict = {}
    for rule in stylesheet.rules:
        this_rule = rule.selector.as_css()
        cssdict[this_rule] = {}
        print (this_rule)
        for css_attr in rule.declarations:
            # print (css_attr)
            cssdict[this_rule][css_attr.name] = css_attr.value.as_css()
            print (css_attr.name, css_attr.value.as_css())
    print(stylesheet.errors)
    return cssdict
    **implementation ends**
    """

    # are we inside /* and */ block?
    inside_comment_block = False

    # are we inside { and  } block?
    reading_values = False

    # main dict to store parsed css file as a dict
    cssdict = {}
    # running css rule value eg. body
    running_css_rule = ""
    # running css attribute values of current css rule
    # eg. attributes inside body
    running_css_attrs_dict = {}

    # iterate through each line in the css content ("\n" delimited)
    for line in _content.splitlines():
        # only after "{" is read
        if reading_values:
            # re-initiate running variabels and stop parsing attributes
            # for the current rule
            if "}" in line:
                assert running_css_rule != ""

                cssdict[running_css_rule] = running_css_attrs_dict

                running_css_rule = ""
                running_css_attrs_dict = {}
                reading_values = False
                continue

            # otherwise, keep adding attribtues to the dict
            # corresponding to the running css rule
            k_v = line.split(":", 2)

            k = k_v[0].strip()
            # handle special case url(https://..)
            v = ":".join(k_v[1:]).strip()[:-1]  # do not include ;

            running_css_attrs_dict[k] = v

        # very naive assumption about comment blocks
        if "/* **" in line and "** */" in line:
            continue

        if "/*" in line:
            inside_comment_block = True
            continue

        if "*/" in line:
            inside_comment_block = False
            continue

        # assumption is made that comment blocks are not found inside
        # { and } blocks
        if inside_comment_block is False:
            # we start parsing attribtues from next line
            if "{" in line:
                running_css_rule = line.split("{")[0].strip()
                assert running_css_attrs_dict == {}

                reading_values = True

    return cssdict

decode_base64()

def ody_migr_utils.decode_base64

(

_encode

)

Decode base64 encoded content to string

Args:
    _encode : bytes-like object

Returns:
    decoded string

Definition at line 45 of file ody_migr_utils.py.

def decode_base64(_encode):
    """Decode base64 encoded content to string

    Args:
        _encode : bytes-like object

    Returns:
        decoded string
    """
    return base64.b64decode(_encode).decode(encoding="utf-8")

encode_base64()

def ody_migr_utils.encode_base64

(

_data

)

Encode content into using base64 encoding

Args:
    _data (str): input data to encode

Returns:
    base64 encoded bytes

Definition at line 33 of file ody_migr_utils.py.

def encode_base64(_data):
    """Encode content into using base64 encoding

    Args:
        _data (str): input data to encode

    Returns:
        base64 encoded bytes
    """
    return base64.b64encode(_data.encode(encoding="utf-8"))

file_exc_decorator()

def ody_migr_utils.file_exc_decorator

(

original_file_handler

)

decorator to handle all file operation exceptions

Definition at line 186 of file ody_migr_utils.py.

def file_exc_decorator(original_file_handler):
    """decorator to handle all file operation exceptions"""
    @wraps(original_file_handler)
    def wrapper(*args, **kwargs):
        try:
            return original_file_handler(*args, **kwargs)
        except (IOError, NameError) as e:
            error_msg = format_error_permission(e, "")
            LOGGER.error(error_msg)
            raise SystemExit(error_msg)
        except Exception as e:
            LOGGER.error(e)
            raise SystemExit(e)
        except:
            # sys.exc_info() returns a tuple (type, value, traceback)
            error_msg = sys.exc_info()[1].args[0]
            LOGGER.error(error_msg)
            raise SystemExit(error_msg)
    return wrapper

format_error_permission()

def ody_migr_utils.format_error_permission	(		_e,
			_file__or_script
	)

Display error messages for FileNotFound/Permission Denied errors.

Args:
    _e:                 error message object
    _file_or_script:    script/file that caused error

Returns:
    Formatted PermissionError/FileNotFoundError error message

Definition at line 163 of file ody_migr_utils.py.

def format_error_permission(_e, _file__or_script):
    """Display error messages for FileNotFound/Permission Denied errors.

    Args:
        _e:                 error message object
        _file_or_script:    script/file that caused error

    Returns:
        Formatted PermissionError/FileNotFoundError error message
    """
    error_msg = ""
    # PermissionError
    if _e.errno == EPERM or _e.errno == EACCES:
        error_msg = "PermissionError - error({}): {} for: {}\t".format(
            _e.errno, _e.strerror, _file__or_script)
    # FileNotFoundError
    elif _e.errno == ENOENT:
        error_msg = "FileNotFoundError - error({}): {} as:{}\t".format(
            _e.errno, _e.strerror, _file__or_script)
    # returning for logging purpose whenever needed
    return error_msg

generate_hash()

def ody_migr_utils.generate_hash	(		_action,
			_cu,
			_secret_key
	)

Generate and return authentication hash code

Args:
    _action:    Data target
    _cu:        Current CU

Returns:
    SHA1 encoded hexdigest code

Definition at line 147 of file ody_migr_utils.py.

def generate_hash(_action, _cu, _secret_key):
    """Generate and return authentication hash code

    Args:
        _action:    Data target
        _cu:        Current CU

    Returns:
        SHA1 encoded hexdigest code
    """
    assert _secret_key is not None
    assert type(_secret_key) == str
    message = "".join([_action, _cu, _secret_key])
    return hashlib.sha1(bytes(message, encoding='utf-8')).hexdigest()

generate_mdx_hash()

def ody_migr_utils.generate_mdx_hash	(		_cu,
			_key
	)

Generate and return MDX new_id hash

Args:
    _cu:        Current CU
    _key:       key to append with _cu

Returns:
    SHA1 encoded hexdigest code

Definition at line 131 of file ody_migr_utils.py.

def generate_mdx_hash(_cu, _key):
    """Generate and return MDX new_id hash

    Args:
        _cu:        Current CU
        _key:       key to append with _cu

    Returns:
        SHA1 encoded hexdigest code
    """
    assert _key is not None
    assert type(_key) == str
    message = "{}{}".format(_cu, _key)
    return hashlib.sha1(bytes(message, encoding='utf-8')).hexdigest()

get_ith_batch()

def ody_migr_utils.get_ith_batch	(		_list,
			batch_index,
			batch_size
	)

Utility function to obtain a ith batch of data from list

Args:
    _list:          list to choose batch from
    batch_index:    batch index
    batch_size:     size of batch to select

Returns:
    batch (sub list of _list) of size batch_size

Definition at line 17 of file ody_migr_utils.py.

def get_ith_batch(_list, batch_index, batch_size):
    """Utility function to obtain a ith batch of data from list

    Args:
        _list:          list to choose batch from
        batch_index:    batch index
        batch_size:     size of batch to select

    Returns:
        batch (sub list of _list) of size batch_size
    """
    start = batch_index * batch_size
    end = min(batch_index * batch_size + batch_size, len(_list))
    return _list[start:end]

get_strip_value_dict()

def ody_migr_utils.get_strip_value_dict	(		_k,
			_dict,
		**	kwargs
	)

Get or Pop an element from a dictionary

Args:
    _k:         key to get or pop
    _dict:      dictionary to get or pop an element from
    **kwargs:   Options:
                default: default value to return
                pop:     do we `get` or `pop` from the dict

Returns:
    value associated with key in a dictionary

Raises:
    ValueError, TypeError if the invalid castto is encountered

Definition at line 207 of file ody_migr_utils.py.

def get_strip_value_dict(_k, _dict, **kwargs):
    """Get or Pop an element from a dictionary

    Args:
        _k:         key to get or pop
        _dict:      dictionary to get or pop an element from
        **kwargs:   Options:
                    default: default value to return
                    pop:     do we `get` or `pop` from the dict

    Returns:
        value associated with key in a dictionary

    Raises:
        ValueError, TypeError if the invalid castto is encountered
    """
    default = kwargs.get("default", None)
    pop = kwargs.get("pop", False)

    if pop:
        val = _dict.pop(_k, default)
    else:
        val = _dict.get(_k, default)

    if val in [None, ''] and default not in [None, '']:
        val = default

    if hasattr(val, 'strip'):
        val = val.strip()

    return val

get_valid_json()

def ody_migr_utils.get_valid_json	(		_obj,
		**	kwargs
	)

Utility function to print pretty json string

Args:
    _obj:       python dictionary
    **kwargs:   optional keywords arguments
                possible values:
                    pretty: for pretty printing
Returns:
    prints formatted json string

Definition at line 57 of file ody_migr_utils.py.

def get_valid_json(_obj, **kwargs):
    """Utility function to print pretty json string

    Args:
        _obj:       python dictionary
        **kwargs:   optional keywords arguments
                    possible values:
                        pretty: for pretty printing
    Returns:
        prints formatted json string

    """
    pretty = kwargs.get("pretty", False)
    if pretty:
        # formatted print layout
        return json.dumps(_obj, indent=4, sort_keys=True,
                          separators=(',', ':'))
    # compact encoding
    return json.dumps(_obj)

log_progress()

def ody_migr_utils.log_progress

(

_progress_dict

)

Log progress information to be used in progressbar in Monitor

Args:
    _progress_dict: dictionary that contains completed and total stat

Definition at line 240 of file ody_migr_utils.py.

def log_progress(_progress_dict):
    """Log progress information to be used in progressbar in Monitor

    Args:
        _progress_dict: dictionary that contains completed and total stat
    """
    LOGGER.info("[PROGRESS] Tables imported: {}/{}".format(
        _progress_dict["completed"], _progress_dict["total"]))

sanitize_dict()

def ody_migr_utils.sanitize_dict

(

records_dict

)

Sanitize dictionary values

Args:
    records_dict: dictionary values

Returns:
    list of stripped, sanitized dictionary

Definition at line 96 of file ody_migr_utils.py.

def sanitize_dict(records_dict):
    """Sanitize dictionary values

    Args:
        records_dict: dictionary values

    Returns:
        list of stripped, sanitized dictionary
    """
    for k in records_dict.keys():
        v = records_dict[k]
        if hasattr(v, 'strip'):
            v = v.strip()
        records_dict[k] = v
    return records_dict

sanitize_insert_collection()

def ody_migr_utils.sanitize_insert_collection

(

_collection

)

Cleanup list values

Args:
    _collection: list of dictionary values

Returns:
    list of stripped, sanitized list of dictionary values

Definition at line 113 of file ody_migr_utils.py.

def sanitize_insert_collection(_collection):
    """Cleanup list values

    Args:
        _collection: list of dictionary values

    Returns:
        list of stripped, sanitized list of dictionary values
    """
    assert type(_collection) == list
    if len(_collection) > 0:
        assert type(_collection[0]) == dict

    for records_dict in _collection:
        records_dict = sanitize_dict(records_dict)
    return _collection

sanitize_insert_values()

def ody_migr_utils.sanitize_insert_values

(

_values

)

Cleanup list values

Args:
    _values: list of values

Returns:
    list of normalized list

Definition at line 78 of file ody_migr_utils.py.

def sanitize_insert_values(_values):
    """Cleanup list values

    Args:
        _values: list of values

    Returns:
        list of normalized list
    """
    assert type(_values) == list
    sanitized_values = []
    for val in _values:
        if hasattr(val, 'strip'):
            val = val.strip()
        sanitized_values.append(val)
    return sanitized_values