benedict.utils.io_util

Complexity

Total Complexity

16

Size/Duplication

Total Lines	84
Duplicated Lines	0 %

Importance

Changes

0

6 Functions

Rating	Name	Duplication	Size	Complexity
A	encode()	0	7	2
A	write_file()	0	14	4
A	read_content()	0	20	5
A	read_url()	0	9	2
A	read_file()	0	5	1
A	decode()	0	8	2

# -*- coding: utf-8 -*-

from benedict.serializers import (
    get_serializer_by_format, get_serializers_extensions, )

import errno
import os
import requests


def decode(s, format, **kwargs):
    serializer = get_serializer_by_format(format)
    if serializer:
        decode_opts = kwargs.copy()
        data = serializer.decode(s.strip(), **decode_opts)
        return data
    else:
        raise ValueError('Invalid format: {}.'.format(format))


def encode(d, format, **kwargs):
    serializer = get_serializer_by_format(format)
    if serializer:
        s = serializer.encode(d, **kwargs)
        return s
    else:
        raise ValueError('Invalid format: {}.'.format(format))


def read_content(s):
    # s -> filepath or url or data
    num_lines = len(s.splitlines())
    if num_lines > 1:
        # data
        return s
    if any([s.startswith(protocol)
            for protocol in ['http://', 'https://']]):
        # url
        return read_url(s)
    elif any([s.endswith(extension)
              for extension in get_serializers_extensions()]):
        # filepath
        if os.path.isfile(s):
            return read_file(s)
        else:
            return None
    else:
        # data
        return s


def read_file(filepath):
    handler = open(filepath, 'r')
    content = handler.read()
    handler.close()
    return content


def read_url(url, *args, **kwargs):
    response = requests.get(url, *args, **kwargs)
    if response.status_code == requests.codes.ok:
        content = response.text
        return content
    else:
        raise ValueError(
            'Invalid url response status code: {}.'.format(
                response.status_code))


def write_file(filepath, content):
    # https://stackoverflow.com/questions/12517451/automatically-creating-directories-with-file-output
    filedir = os.path.dirname(filepath)
    if not os.path.exists(filedir):
        try:
            os.makedirs(filedir)
        except OSError as e:
            # Guard against race condition
            if e.errno != errno.EEXIST:
                raise e
    handler = open(filepath, 'w+')
    handler.write(content)
    handler.close()
    return True