Improved csv I/O support and refacored io dict and utils.

This commit is contained in:
Fabio Caccamo 2019-11-07 17:45:45 +01:00
parent 2f1f680a83
commit b61eb4ea44
7 changed files with 480 additions and 181 deletions

View File

@ -52,8 +52,13 @@ class benedict(IODict, KeypathDict, ParseDict):
@staticmethod
@benediction
def from_base64(s, format='json', **kwargs):
return IODict.from_base64(s, format, **kwargs)
def from_base64(s, subformat='json', encoding='utf-8', **kwargs):
return IODict.from_base64(s, subformat=subformat, encoding=encoding, **kwargs)
@staticmethod
@benediction
def from_csv(s, columns=None, columns_row=True, **kwargs):
return IODict.from_csv(s, columns=columns, columns_row=columns_row, **kwargs)
@staticmethod
@benediction

View File

@ -8,30 +8,34 @@ from six import string_types, text_type
class IODict(dict):
def __init__(self, *args, **kwargs):
# if first argument is data-string,
# try to decode it using all decoders.
# if first argument is data-string try to decode it.
# use 'format' kwarg to specify the decoder to use, default 'json'.
if len(args) and isinstance(args[0], string_types):
d = IODict._from_any_data_string(args[0], **kwargs)
s = args[0]
format = kwargs.pop('format', 'json').lower()
if format in ['b64', 'base64']:
kwargs.setdefault('subformat', 'json')
# decode data-string and initialize with dict data.
d = IODict._decode(s, format, **kwargs)
if d and isinstance(d, dict):
args = list(args)
args[0] = d
args = tuple(args)
super(IODict, self).__init__(d)
else:
raise ValueError('Invalid string data input.')
super(IODict, self).__init__(*args, **kwargs)
else:
super(IODict, self).__init__(*args, **kwargs)
@staticmethod
def _decode(s, decoder, **kwargs):
def _decode(s, format, **kwargs):
d = None
try:
content = io_util.read_content(s)
# decode content using the given decoder
data = decoder(content, **kwargs)
# decode content using the given format
data = io_util.decode(content, format, **kwargs)
if isinstance(data, dict):
d = data
elif isinstance(data, list):
# force list to dict
d = { 'values':data }
d = { 'values': data }
else:
raise ValueError(
'Invalid data type: {}, expected dict or list.'.format(type(data)))
@ -41,90 +45,66 @@ class IODict(dict):
return d
@staticmethod
def _encode(d, encoder, filepath=None, **kwargs):
s = encoder(d, **kwargs)
def _encode(d, format, **kwargs):
filepath = kwargs.pop('filepath', None)
s = io_util.encode(d, format, **kwargs)
if filepath:
io_util.write_file(filepath, s)
return s
@staticmethod
def _from_any_data_string(s, **kwargs):
funcs = [
IODict.from_base64,
IODict.from_json,
IODict.from_query_string,
IODict.from_toml,
IODict.from_xml,
IODict.from_yaml,
]
for f in funcs:
try:
options = kwargs.copy()
d = f(s, **options)
return d
except ValueError:
pass
def from_base64(s, subformat='json', encoding='utf-8', **kwargs):
kwargs['subformat'] = subformat
kwargs['encoding'] = encoding
return IODict._decode(s, 'base64', **kwargs)
@staticmethod
def from_base64(s, format='json', encoding='utf-8', **kwargs):
kwargs['format'] = format
kwargs['encoding'] = encoding
return IODict._decode(s,
decoder=io_util.decode_base64, **kwargs)
def from_csv(s, columns=None, columns_row=True, **kwargs):
kwargs['columns'] = columns
kwargs['columns_row'] = columns_row
return IODict._decode(s, 'csv', **kwargs)
@staticmethod
def from_json(s, **kwargs):
return IODict._decode(s,
decoder=io_util.decode_json, **kwargs)
return IODict._decode(s, 'json', **kwargs)
@staticmethod
def from_query_string(s, **kwargs):
return IODict._decode(s,
decoder=io_util.decode_query_string, **kwargs)
return IODict._decode(s, 'query_string', **kwargs)
@staticmethod
def from_toml(s, **kwargs):
return IODict._decode(s,
decoder=io_util.decode_toml, **kwargs)
return IODict._decode(s, 'toml', **kwargs)
@staticmethod
def from_xml(s, **kwargs):
return IODict._decode(s,
decoder=io_util.decode_xml, **kwargs)
return IODict._decode(s, 'xml', **kwargs)
@staticmethod
def from_yaml(s, **kwargs):
return IODict._decode(s,
decoder=io_util.decode_yaml, **kwargs)
return IODict._decode(s, 'yaml', **kwargs)
def to_base64(self, filepath=None, format='json', encoding='utf-8', **kwargs):
kwargs['format'] = format
def to_base64(self, subformat='json', encoding='utf-8', **kwargs):
kwargs['subformat'] = subformat
kwargs['encoding'] = encoding
return IODict._encode(self,
encoder=io_util.encode_base64,
filepath=filepath, **kwargs)
return IODict._encode(self, 'base64', **kwargs)
def to_json(self, filepath=None, **kwargs):
return IODict._encode(self,
encoder=io_util.encode_json,
filepath=filepath, **kwargs)
def to_csv(self, key='values', columns=None, columns_row=True, **kwargs):
kwargs['columns'] = columns
kwargs['columns_row'] = columns_row
return IODict._encode(self[key], 'csv', **kwargs)
def to_query_string(self, filepath=None, **kwargs):
return IODict._encode(self,
encoder=io_util.encode_query_string,
filepath=filepath, **kwargs)
def to_json(self, **kwargs):
return IODict._encode(self, 'json', **kwargs)
def to_toml(self, filepath=None, **kwargs):
return IODict._encode(self,
encoder=io_util.encode_toml,
filepath=filepath, **kwargs)
def to_query_string(self, **kwargs):
return IODict._encode(self, 'query_string', **kwargs)
def to_xml(self, filepath=None, **kwargs):
return IODict._encode(self,
encoder=io_util.encode_xml,
filepath=filepath, **kwargs)
def to_toml(self, **kwargs):
return IODict._encode(self, 'toml', **kwargs)
def to_yaml(self, filepath=None, **kwargs):
return IODict._encode(self,
encoder=io_util.encode_yaml,
filepath=filepath, **kwargs)
def to_xml(self, **kwargs):
return IODict._encode(self, 'xml', **kwargs)
def to_yaml(self, **kwargs):
return IODict._encode(self, 'yaml', **kwargs)

View File

@ -1,6 +1,7 @@
# -*- coding: utf-8 -*-
from six import binary_type, string_types, StringIO
from slugify import slugify
import base64
import csv
@ -27,6 +28,16 @@ except ImportError:
from urlparse import parse_qs
def decode(s, format, **kwargs):
decode_func = _get_format_decoder(format)
if decode_func:
decode_opts = kwargs.copy()
data = decode_func(s.strip(), **decode_opts)
return data
else:
raise ValueError('Invalid format: {}.'.format(format))
def decode_base64(s, **kwargs):
# fix urlencoded chars
s = unquote(s)
@ -35,18 +46,12 @@ def decode_base64(s, **kwargs):
if m != 0:
s += '=' * (4 - m)
data = base64.b64decode(s)
format = kwargs.pop('format', None)
encoding = kwargs.pop('encoding', 'utf-8' if format else None)
subformat = kwargs.pop('subformat', None)
encoding = kwargs.pop('encoding', 'utf-8' if subformat else None)
if encoding:
data = data.decode(encoding)
if format:
decoders = {
'json': decode_json,
'toml': decode_toml,
'yaml': decode_yaml,
'xml': decode_xml,
}
decode_func = decoders.get(format.lower(), '')
if subformat:
decode_func = _get_format_decoder(subformat)
if decode_func:
data = decode_func(data, **kwargs)
return data
@ -109,18 +114,21 @@ def decode_yaml(s, **kwargs):
return data
def encode(d, format, **kwargs):
encode_func = _get_format_encoder(format)
if encode_func:
s = encode_func(d, **kwargs)
return s
else:
raise ValueError('Invalid format: {}.'.format(format))
def encode_base64(d, **kwargs):
data = d
format = kwargs.pop('format', None)
encoding = kwargs.pop('encoding', 'utf-8' if format else None)
if not isinstance(data, string_types) and format:
encoders = {
'json': encode_json,
'toml': encode_toml,
'yaml': encode_yaml,
'xml': encode_xml,
}
encode_func = encoders.get(format.lower(), '')
subformat = kwargs.pop('subformat', None)
encoding = kwargs.pop('encoding', 'utf-8' if subformat else None)
if not isinstance(data, string_types) and subformat:
encode_func = _get_format_encoder(subformat)
if encode_func:
data = encode_func(data, **kwargs)
if isinstance(data, string_types) and encoding:
@ -225,3 +233,60 @@ def write_file(filepath, content):
handler.write(content)
handler.close()
return True
_formats = {
'b64': {
'decoder': decode_base64,
'encoder': encode_base64,
},
'base64': {
'decoder': decode_base64,
'encoder': encode_base64,
},
'csv': {
'decoder': decode_csv,
'encoder': encode_csv,
},
'json': {
'decoder': decode_json,
'encoder': encode_json,
},
'qs': {
'decoder': decode_query_string,
'encoder': encode_query_string,
},
'query_string': {
'decoder': decode_query_string,
'encoder': encode_query_string,
},
'toml': {
'decoder': decode_toml,
'encoder': encode_toml,
},
'yaml': {
'decoder': decode_yaml,
'encoder': encode_yaml,
},
'yml': {
'decoder': decode_yaml,
'encoder': encode_yaml,
},
'xml': {
'decoder': decode_xml,
'encoder': encode_xml,
},
}
def _get_format(format):
return _formats.get(
slugify(format, separator='_'), {})
def _get_format_decoder(format):
return _get_format(format).get('decoder', None)
def _get_format_encoder(format):
return _get_format(format).get('encoder', None)

View File

@ -0,0 +1,2 @@
Lorem ipsum consectetur sint id aute officia sed excepteur consectetur labore laboris dolore in labore consequat ut in eu ut deserunt.
Elit aliqua velit aliquip voluptate consequat reprehenderit occaecat dolor ut esse aute laboris cillum fugiat esse est laborum.
1 Lorem ipsum consectetur sint id aute officia sed excepteur consectetur labore laboris dolore in labore consequat ut in eu ut deserunt.
2 Elit aliqua velit aliquip voluptate consequat reprehenderit occaecat dolor ut esse aute laboris cillum fugiat esse est laborum.

View File

@ -0,0 +1,5 @@
id,name,age,height,weight
1,Alice,20,62,120.6
2,Freddie,21,74,190.6
3,Bob,17,68,120.0
4,François,32,75,110.05
1 id name age height weight
2 1 Alice 20 62 120.6
3 2 Freddie 21 74 190.6
4 3 Bob 17 68 120.0
5 4 François 32 75 110.05

View File

@ -494,10 +494,42 @@ class BenedictTestCase(unittest.TestCase):
d = benedict.from_base64(j)
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
# constructor
d = benedict(j)
# static method with subformat
d = benedict.from_base64(j, subformat='json')
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
# constructor
d = benedict(j, format='base64')
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
# constructor with subformat
d = benedict(j, format='base64', subformat='json')
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
def test_from_csv_with_valid_data(self):
s = """id,name,age,height,weight
1,Alice,20,62,120.6
2,Freddie,21,74,190.6
3,Bob,17,68,120.0
4,François,32,75,110.05
"""
r = {
'values': [
{ 'id':'1', 'name':'Alice', 'age':'20', 'height':'62', 'weight':'120.6', },
{ 'id':'2', 'name':'Freddie', 'age':'21', 'height':'74', 'weight':'190.6', },
{ 'id':'3', 'name':'Bob', 'age':'17', 'height':'68', 'weight':'120.0', },
{ 'id':'4', 'name':'François', 'age':'32', 'height':'75', 'weight':'110.05', },
],
}
# static method
d = benedict.from_csv(s)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
# constructor
d = benedict(s, format='csv')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
def test_from_json(self):
j = '{"a": 1, "b": 2, "c": 3}'
@ -506,7 +538,7 @@ class BenedictTestCase(unittest.TestCase):
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
# constructor
d = benedict(j)
d = benedict(j, format='json')
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
@ -518,7 +550,7 @@ class BenedictTestCase(unittest.TestCase):
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, r)
# constructor
d = benedict(s)
d = benedict(s, format='query-string')
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, r)
@ -535,42 +567,43 @@ class BenedictTestCase(unittest.TestCase):
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a':1, 'b':{ 'c':3, 'd':4 },})
# constructor
d = benedict(j)
d = benedict(j, format='toml')
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a':1, 'b':{ 'c':3, 'd':4 },})
def test_from_xml(self):
j = """<?xml version="1.0" ?>
<root>
<a>1</a>
<b>
<c>3</c>
<d>4</d>
</b>
</root>
"""
j = """
<?xml version="1.0" ?>
<root>
<a>1</a>
<b>
<c>3</c>
<d>4</d>
</b>
</root>
"""
# static method
d = benedict.from_xml(j)
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d.get('root'), { 'a':'1', 'b':{ 'c':'3', 'd':'4' },})
# constructor
d = benedict(j)
d = benedict(j, format='xml')
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d.get('root'), { 'a':'1', 'b':{ 'c':'3', 'd':'4' },})
def test_from_yaml(self):
j = """
a: 1
b:
c: 3
d: 4
"""
a: 1
b:
c: 3
d: 4
"""
# static method
d = benedict.from_yaml(j)
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a':1, 'b':{ 'c':3, 'd':4 },})
# constructor
d = benedict(j)
d = benedict(j, format='yaml')
self.assertTrue(isinstance(d, benedict))
self.assertEqual(d, { 'a':1, 'b':{ 'c':3, 'd':4 },})

View File

@ -33,7 +33,11 @@ class IODictTestCase(unittest.TestCase):
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
# constructor
d = IODict(j)
d = IODict(j, format='base64')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
# constructor with subformat
d = IODict(j, format='base64', subformat='json')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
@ -46,7 +50,7 @@ class IODictTestCase(unittest.TestCase):
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, 'd': 4})
# constructor
d = IODict(j)
d = IODict(j, format='base64')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, 'd': 4})
@ -57,7 +61,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_base64(j)
# constructor
with self.assertRaises(ValueError):
IODict(j)
IODict(j, format='base64')
def test_from_base64_with_valid_file_valid_content(self):
filepath = self.input_path('valid-content.base64')
@ -65,7 +69,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_base64(filepath)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(filepath)
d = IODict(filepath, format='base64')
self.assertTrue(isinstance(d, dict))
def test_from_base64_with_valid_file_valid_content_invalid_format(self):
@ -92,7 +96,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_base64(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='base64')
def test_from_base64_with_invalid_file(self):
filepath = self.input_path('invalid-file.base64')
@ -101,7 +105,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_base64(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='base64')
def test_from_base64_with_valid_url_valid_content(self):
url = 'https://raw.githubusercontent.com/fabiocaccamo/python-benedict/master/tests/input/valid-content.base64'
@ -109,7 +113,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_base64(url)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(url)
d = IODict(url, format='base64')
self.assertTrue(isinstance(d, dict))
def test_from_base64_with_valid_url_invalid_content(self):
@ -119,7 +123,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_base64(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='base64')
def test_from_base64_with_invalid_url(self):
url = 'https://github.com/fabiocaccamo/python-benedict-invalid'
@ -128,7 +132,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_base64(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='base64')
def test_to_base64(self):
d = IODict({
@ -150,6 +154,210 @@ class IODictTestCase(unittest.TestCase):
self.assertTrue(d, os.path.isfile(filepath))
self.assertEqual(d, IODict.from_base64(filepath))
# CSV
def test_from_csv_with_valid_data(self):
s = """id,name,age,height,weight
1,Alice,20,62,120.6
2,Freddie,21,74,190.6
3,Bob,17,68,120.0
4,François,32,75,110.05
"""
r = {
'values': [
{ 'id':'1', 'name':'Alice', 'age':'20', 'height':'62', 'weight':'120.6', },
{ 'id':'2', 'name':'Freddie', 'age':'21', 'height':'74', 'weight':'190.6', },
{ 'id':'3', 'name':'Bob', 'age':'17', 'height':'68', 'weight':'120.0', },
{ 'id':'4', 'name':'François', 'age':'32', 'height':'75', 'weight':'110.05', },
],
}
# static method
d = IODict.from_csv(s)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
# constructor
d = IODict(s, format='csv')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
# def test_from_csv_with_invalid_data(self):
# s = 'Lorem ipsum est in ea occaecat nisi officia.'
# # static method
# with self.assertRaises(ValueError):
# print(IODict.from_csv(s))
# # constructor
# with self.assertRaises(ValueError):
# IODict(s, format='csv')
def test_from_csv_with_valid_file_valid_content(self):
filepath = self.input_path('valid-content.csv')
# static method
d = IODict.from_csv(filepath)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(filepath, format='csv')
self.assertTrue(isinstance(d, dict))
# def test_from_csv_with_valid_file_valid_content_invalid_format(self):
# filepath = self.input_path('valid-content.base64')
# with self.assertRaises(ValueError):
# IODict.from_csv(filepath)
# filepath = self.input_path('valid-content.qs')
# with self.assertRaises(ValueError):
# IODict.from_csv(filepath)
# filepath = self.input_path('valid-content.toml')
# with self.assertRaises(ValueError):
# IODict.from_csv(filepath)
# filepath = self.input_path('valid-content.xml')
# with self.assertRaises(ValueError):
# IODict.from_csv(filepath)
# filepath = self.input_path('valid-content.yml')
# with self.assertRaises(ValueError):
# IODict.from_csv(filepath)
# def test_from_csv_with_valid_file_invalid_content(self):
# filepath = self.input_path('invalid-content.csv')
# # static method
# with self.assertRaises(ValueError):
# IODict.from_csv(filepath)
# # constructor
# with self.assertRaises(ValueError):
# IODict(filepath, format='csv')
# def test_from_csv_with_invalid_file(self):
# filepath = self.input_path('invalid-file.csv')
# # static method
# with self.assertRaises(ValueError):
# IODict.from_csv(filepath)
# # constructor
# with self.assertRaises(ValueError):
# IODict(filepath, format='csv')
# def test_from_csv_with_valid_url_valid_content(self):
# url = 'https://raw.githubusercontent.com/fabiocaccamo/python-benedict/master/tests/input/valid-content.csv'
# # static method
# d = IODict.from_csv(url)
# self.assertTrue(isinstance(d, dict))
# # constructor
# d = IODict(url, format='csv')
# self.assertTrue(isinstance(d, dict))
# def test_from_csv_with_valid_url_invalid_content(self):
# url = 'https://github.com/fabiocaccamo/python-benedict'
# # static method
# with self.assertRaises(ValueError):
# IODict.from_csv(url)
# # constructor
# with self.assertRaises(ValueError):
# IODict(url, format='csv')
def test_from_csv_with_invalid_url(self):
url = 'https://github.com/fabiocaccamo/python-benedict-invalid'
# static method
with self.assertRaises(ValueError):
IODict.from_csv(url)
# constructor
with self.assertRaises(ValueError):
IODict(url, format='csv')
def test_to_csv(self):
d = IODict({
'values': [
{ 'id':'1', 'name':'Alice', 'age':'20', 'height':'62', 'weight':'120.6', },
{ 'id':'2', 'name':'Freddie', 'age':'21', 'height':'74', 'weight':'190.6', },
{ 'id':'3', 'name':'Bob', 'age':'17', 'height':'68', 'weight':'120.0', },
{ 'id':'4', 'name':'François', 'age':'32', 'height':'75', 'weight':'110.05', },
],
})
s = d.to_csv()
r = """age,height,id,name,weight
20,62,1,Alice,120.6
21,74,2,Freddie,190.6
17,68,3,Bob,120.0
32,75,4,François,110.05
"""
self.assertEqual(s, r)
def test_to_csv_with_custom_columns(self):
d = IODict({
'values': [
{ 'id':'1', 'name':'Alice', 'age':'20', 'height':'62', 'weight':'120.6', },
{ 'id':'2', 'name':'Freddie', 'age':'21', 'height':'74', 'weight':'190.6', },
{ 'id':'3', 'name':'Bob', 'age':'17', 'height':'68', 'weight':'120.0', },
{ 'id':'4', 'name':'François', 'age':'32', 'height':'75', 'weight':'110.05', },
],
})
s = d.to_csv(key='values', columns=['id', 'name', 'family_name', 'age', 'height', 'gender', 'weight'])
r = """id,name,family_name,age,height,gender,weight
1,Alice,,20,62,,120.6
2,Freddie,,21,74,,190.6
3,Bob,,17,68,,120.0
4,François,,32,75,,110.05
"""
self.assertEqual(s, r)
def test_to_csv_with_custom_delimiter_and_quotes(self):
d = IODict({
'values': [
{ 'id':'1', 'name':'Alice', 'age':'20', 'height':'62', 'weight':'120.6', },
{ 'id':'2', 'name':'Freddie', 'age':'21', 'height':'74', 'weight':'190.6', },
{ 'id':'3', 'name':'Bob', 'age':'17', 'height':'68', 'weight':'120.0', },
{ 'id':'4', 'name':'François', 'age':'32', 'height':'75', 'weight':'110.05', },
],
})
s = d.to_csv(columns=['id', 'name', 'age', 'height', 'weight'], delimiter=";", quote=True)
r = """"id";"name";"age";"height";"weight"
"1";"Alice";"20";"62";"120.6"
"2";"Freddie";"21";"74";"190.6"
"3";"Bob";"17";"68";"120.0"
"4";"François";"32";"75";"110.05"
"""
self.assertEqual(s, r)
def test_to_csv_with_custom_key_valid(self):
d = IODict({
'results': [
{ 'id':'1', 'name':'Alice', 'age':'20', 'height':'62', 'weight':'120.6', },
{ 'id':'2', 'name':'Freddie', 'age':'21', 'height':'74', 'weight':'190.6', },
{ 'id':'3', 'name':'Bob', 'age':'17', 'height':'68', 'weight':'120.0', },
{ 'id':'4', 'name':'François', 'age':'32', 'height':'75', 'weight':'110.05', },
],
})
s = d.to_csv('results', columns=['id', 'name', 'age', 'height', 'weight'])
r = """id,name,age,height,weight
1,Alice,20,62,120.6
2,Freddie,21,74,190.6
3,Bob,17,68,120.0
4,François,32,75,110.05
"""
self.assertEqual(s, r)
def test_to_csv_with_custom_key_invalid(self):
d = IODict({
'values': [
{ 'id':'1', 'name':'Alice', 'age':'20', 'height':'62', 'weight':'120.6', },
{ 'id':'2', 'name':'Freddie', 'age':'21', 'height':'74', 'weight':'190.6', },
{ 'id':'3', 'name':'Bob', 'age':'17', 'height':'68', 'weight':'120.0', },
{ 'id':'4', 'name':'François', 'age':'32', 'height':'75', 'weight':'110.05', },
],
})
with self.assertRaises(KeyError):
s = d.to_csv('invalid_values', columns=['id', 'name', 'age', 'height', 'weight'])
def test_to_csv_file(self):
d = IODict({
'values': [
{ 'id':'1', 'name':'Alice', 'age':'20', 'height':'62', 'weight':'120.6', },
{ 'id':'2', 'name':'Freddie', 'age':'21', 'height':'74', 'weight':'190.6', },
{ 'id':'3', 'name':'Bob', 'age':'17', 'height':'68', 'weight':'120.0', },
{ 'id':'4', 'name':'François', 'age':'32', 'height':'75', 'weight':'110.05', },
],
})
filepath = self.output_path('test_to_csv_file.csv')
d.to_csv(filepath=filepath)
self.assertTrue(d, os.path.isfile(filepath))
self.assertEqual(d, IODict.from_csv(filepath))
# JSON
def test_from_json_with_valid_data(self):
@ -159,7 +367,7 @@ class IODictTestCase(unittest.TestCase):
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
# constructor
d = IODict(j)
d = IODict(j, format='json')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
@ -170,7 +378,7 @@ class IODictTestCase(unittest.TestCase):
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'values': [0, 1, 2, 3, 4, 5] })
# constructor
d = IODict(j)
d = IODict(j, format='json')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'values': [0, 1, 2, 3, 4, 5] })
@ -202,7 +410,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_json(j)
# constructor
with self.assertRaises(ValueError):
IODict(j)
IODict(j, format='json')
def test_from_json_with_valid_file_valid_content(self):
filepath = self.input_path('valid-content.json')
@ -210,7 +418,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_json(filepath)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(filepath)
d = IODict(filepath, format='json')
self.assertTrue(isinstance(d, dict))
def test_from_json_with_valid_file_valid_content_invalid_format(self):
@ -237,7 +445,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_json(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='json')
def test_from_json_with_invalid_file(self):
filepath = self.input_path('invalid-file.json')
@ -246,7 +454,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_json(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='json')
def test_from_json_with_valid_url_valid_content(self):
url = 'https://raw.githubusercontent.com/fabiocaccamo/python-benedict/master/tests/input/valid-content.json'
@ -254,7 +462,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_json(url)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(url)
d = IODict(url, format='json')
self.assertTrue(isinstance(d, dict))
def test_from_json_with_valid_url_invalid_content(self):
@ -264,7 +472,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_json(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='json')
def test_from_json_with_invalid_url(self):
url = 'https://github.com/fabiocaccamo/python-benedict-invalid'
@ -273,7 +481,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_json(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='json')
def test_to_json(self):
d = IODict({
@ -311,7 +519,7 @@ class IODictTestCase(unittest.TestCase):
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
# constructor
d = IODict(s)
d = IODict(s, format='query_string')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, r)
@ -322,7 +530,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_query_string(s)
# constructor
with self.assertRaises(ValueError):
IODict(s)
IODict(s, format='query_string')
def test_from_query_string_with_valid_file_valid_content(self):
filepath = self.input_path('valid-content.qs')
@ -330,7 +538,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_query_string(filepath)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(filepath)
d = IODict(filepath, format='query_string')
self.assertTrue(isinstance(d, dict))
def test_from_query_string_with_valid_file_valid_content_invalid_format(self):
@ -357,7 +565,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_query_string(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='query_string')
def test_from_query_string_with_invalid_file(self):
filepath = self.input_path('invalid-file.qs')
@ -366,7 +574,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_query_string(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='query_string')
def test_from_query_string_with_valid_url_valid_content(self):
url = 'https://raw.githubusercontent.com/fabiocaccamo/python-benedict/master/tests/input/valid-content.qs'
@ -374,7 +582,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_query_string(url)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(url)
d = IODict(url, format='query_string')
self.assertTrue(isinstance(d, dict))
def test_from_query_string_with_valid_url_invalid_content(self):
@ -384,7 +592,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_query_string(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='query_string')
def test_from_query_string_with_invalid_url(self):
url = 'https://github.com/fabiocaccamo/python-benedict-invalid'
@ -393,7 +601,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_query_string(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='query_string')
def test_to_query_string(self):
data = { 'ok': '1', 'test': '2', 'page': '3', 'lib':'python benedict', 'author':'Fabio Caccamo' }
@ -412,18 +620,18 @@ class IODictTestCase(unittest.TestCase):
def test_from_toml_with_valid_data(self):
j = """
a = 1
a = 1
[b]
c = 3
d = 4
"""
[b]
c = 3
d = 4
"""
# static method
d = IODict.from_toml(j)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a':1, 'b':{ 'c':3, 'd':4 },})
# constructor
d = IODict(j)
d = IODict(j, format='toml')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a':1, 'b':{ 'c':3, 'd':4 },})
@ -434,7 +642,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_toml(j)
# constructor
with self.assertRaises(ValueError):
IODict(j)
IODict(j, format='toml')
def test_from_toml_with_valid_file_valid_content(self):
filepath = self.input_path('valid-content.toml')
@ -442,7 +650,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_toml(filepath)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(filepath)
d = IODict(filepath, format='toml')
self.assertTrue(isinstance(d, dict))
def test_from_toml_with_valid_file_valid_content_invalid_format(self):
@ -469,7 +677,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_toml(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='toml')
def test_from_toml_with_invalid_file(self):
filepath = self.input_path('invalid-file.toml')
@ -478,7 +686,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_toml(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='toml')
def test_from_toml_with_valid_url_valid_content(self):
url = 'https://raw.githubusercontent.com/fabiocaccamo/python-benedict/master/tests/input/valid-content.toml'
@ -486,7 +694,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_toml(url)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(url)
d = IODict(url, format='toml')
self.assertTrue(isinstance(d, dict))
def test_from_toml_with_valid_url_invalid_content(self):
@ -496,7 +704,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_toml(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='toml')
def test_from_toml_with_invalid_url(self):
url = 'https://github.com/fabiocaccamo/python-benedict-invalid'
@ -505,7 +713,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_toml(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='toml')
def test_to_toml(self):
d = IODict({
@ -536,21 +744,22 @@ class IODictTestCase(unittest.TestCase):
# XML
def test_from_xml_with_valid_data(self):
j = """<?xml version="1.0" ?>
<root>
<a>1</a>
<b>
<c>3</c>
<d>4</d>
</b>
</root>
"""
j = """
<?xml version="1.0" ?>
<root>
<a>1</a>
<b>
<c>3</c>
<d>4</d>
</b>
</root>
"""
# static method
d = IODict.from_xml(j)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d.get('root'), { 'a':'1', 'b':{ 'c':'3', 'd':'4' },})
# constructor
d = IODict(j)
d = IODict(j, format='xml')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d.get('root'), { 'a':'1', 'b':{ 'c':'3', 'd':'4' },})
@ -561,7 +770,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_xml(j)
# constructor
with self.assertRaises(ValueError):
IODict(j)
IODict(j, format='xml')
def test_from_xml_with_valid_file_valid_content(self):
filepath = self.input_path('valid-content.xml')
@ -569,7 +778,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_xml(filepath)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(filepath)
d = IODict(filepath, format='xml')
self.assertTrue(isinstance(d, dict))
def test_from_xml_with_valid_file_valid_content_invalid_format(self):
@ -596,7 +805,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_xml(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='xml')
def test_from_xml_with_invalid_file(self):
filepath = self.input_path('invalid-file.xml')
@ -605,7 +814,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_xml(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='xml')
def test_from_xml_with_valid_url_valid_content(self):
url = 'https://raw.githubusercontent.com/fabiocaccamo/python-benedict/master/tests/input/valid-content.xml'
@ -613,7 +822,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_xml(url)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(url)
d = IODict(url, format='xml')
self.assertTrue(isinstance(d, dict))
def test_from_xml_with_valid_url_invalid_content(self):
@ -623,7 +832,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_xml(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='xml')
def test_from_xml_with_invalid_url(self):
url = 'https://github.com/fabiocaccamo/python-benedict-invalid'
@ -632,7 +841,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_xml(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='xml')
def test_to_xml(self):
d = IODict({
@ -668,17 +877,17 @@ class IODictTestCase(unittest.TestCase):
def test_from_yaml_with_valid_data(self):
j = """
a: 1
b:
c: 3
d: 4
"""
a: 1
b:
c: 3
d: 4
"""
# static method
d = IODict.from_yaml(j)
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a':1, 'b':{ 'c':3, 'd':4 },})
# constructor
d = IODict(j)
d = IODict(j, format='yaml')
self.assertTrue(isinstance(d, dict))
self.assertEqual(d, { 'a':1, 'b':{ 'c':3, 'd':4 },})
@ -689,7 +898,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_yaml(j)
# constructor
with self.assertRaises(ValueError):
IODict(j)
IODict(j, format='yaml')
def test_from_yaml_with_valid_file_valid_content(self):
filepath = self.input_path('valid-content.yml')
@ -697,7 +906,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_yaml(filepath)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(filepath)
d = IODict(filepath, format='yaml')
self.assertTrue(isinstance(d, dict))
def test_from_yaml_with_valid_file_valid_content_invalid_format(self):
@ -724,7 +933,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_yaml(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='yaml')
def test_from_yaml_with_invalid_file(self):
filepath = self.input_path('invalid-file.yml')
@ -733,7 +942,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_yaml(filepath)
# constructor
with self.assertRaises(ValueError):
IODict(filepath)
IODict(filepath, format='yaml')
def test_from_yaml_with_valid_url_valid_content(self):
url = 'https://raw.githubusercontent.com/fabiocaccamo/python-benedict/master/tests/input/valid-content.yml'
@ -741,7 +950,7 @@ class IODictTestCase(unittest.TestCase):
d = IODict.from_yaml(url)
self.assertTrue(isinstance(d, dict))
# constructor
d = IODict(url)
d = IODict(url, format='yaml')
self.assertTrue(isinstance(d, dict))
def test_from_yaml_with_valid_url_invalid_content(self):
@ -751,7 +960,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_yaml(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='yaml')
def test_from_yaml_with_invalid_url(self):
url = 'https://github.com/fabiocaccamo/python-benedict-invalid'
@ -760,7 +969,7 @@ class IODictTestCase(unittest.TestCase):
IODict.from_yaml(url)
# constructor
with self.assertRaises(ValueError):
IODict(url)
IODict(url, format='yaml')
def test_to_yaml(self):
d = IODict({