优化覆盖率

This commit is contained in:
JrD
2020-09-03 16:31:27 +08:00
parent 6510a09c63
commit 7ba73dfb9f
3 changed files with 28 additions and 312 deletions
+27 -61
View File
@@ -4,7 +4,33 @@ import csv
from io import StringIO
from uuid import UUID
from . import tablib
""" Tablib - formats
"""
from collections import OrderedDict
class Registry(object):
_formats = OrderedDict()
def register(self, key, format_or_path):
# Create Databook.<format> read or read/write properties
# Create Dataset.<format> read or read/write properties,
# and Dataset.get_<format>/set_<format> methods.
self._formats[key] = format_or_path
def register_builtins(self):
# Registration ordering matters for autodetection.
self.register('csv', CSVFormat())
self.register('json', JSONFormat())
def get_format(self, key):
if key not in self._formats:
raise Exception("OneForAll has no format '%s'." % key)
return self._formats[key]
registry = Registry()
def serialize_objects_handler(obj):
@@ -30,38 +56,6 @@ class JSONFormat(object):
"""Returns JSON representation of Dataset."""
return json.dumps(dataset.dict, default=serialize_objects_handler)
@classmethod
def export_book(cls, databook):
"""Returns JSON representation of Databook."""
return json.dumps(databook._package(), default=serialize_objects_handler)
@classmethod
def import_set(cls, dset, in_stream):
"""Returns dataset from JSON stream."""
dset.wipe()
dset.dict = json.load(in_stream)
@classmethod
def import_book(cls, dbook, in_stream):
"""Returns databook from JSON stream."""
dbook.wipe()
for sheet in json.load(in_stream):
data = tablib.Dataset()
data.title = sheet['title']
data.dict = sheet['data']
dbook.add_sheet(data)
@classmethod
def detect(cls, stream):
"""Returns True if given stream is valid JSON."""
try:
json.load(stream)
return True
except (TypeError, ValueError):
return False
""" Tablib - CSV Support.
"""
@@ -93,31 +87,3 @@ class CSVFormat(object):
"""Returns CSV representation of Dataset."""
stream = cls.export_stream_set(dataset, **kwargs)
return stream.getvalue()
@classmethod
def import_set(cls, dset, in_stream, headers=True, **kwargs):
"""Returns dataset from CSV stream."""
dset.wipe()
kwargs.setdefault('delimiter', cls.DEFAULT_DELIMITER)
rows = csv.reader(in_stream, **kwargs)
for i, row in enumerate(rows):
if (i == 0) and (headers):
dset.headers = row
elif row:
if i > 0 and len(row) < dset.width:
row += [''] * (dset.width - len(row))
dset.append(row)
@classmethod
def detect(cls, stream, delimiter=None):
"""Returns True if given stream is valid CSV."""
try:
csv.Sniffer().sniff(stream.read(1024),
delimiters=delimiter or cls.DEFAULT_DELIMITER)
return True
except Exception:
return False