Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/tablib/formats/_csv.py: 30%
40 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1""" Tablib - *SV Support.
2"""
4import csv
5from io import StringIO
8class CSVFormat:
9 title = 'csv'
10 extensions = ('csv',)
12 DEFAULT_DELIMITER = ','
14 @classmethod
15 def export_stream_set(cls, dataset, **kwargs):
16 """Returns CSV representation of Dataset as file-like."""
17 stream = StringIO()
19 kwargs.setdefault('delimiter', cls.DEFAULT_DELIMITER)
21 _csv = csv.writer(stream, **kwargs)
23 for row in dataset._package(dicts=False):
24 _csv.writerow(row)
26 stream.seek(0)
27 return stream
29 @classmethod
30 def export_set(cls, dataset, **kwargs):
31 """Returns CSV representation of Dataset."""
32 stream = cls.export_stream_set(dataset, **kwargs)
33 return stream.getvalue()
35 @classmethod
36 def import_set(cls, dset, in_stream, headers=True, skip_lines=0, **kwargs):
37 """Returns dataset from CSV stream."""
39 dset.wipe()
41 kwargs.setdefault('delimiter', cls.DEFAULT_DELIMITER)
43 rows = csv.reader(in_stream, **kwargs)
44 for i, row in enumerate(rows):
45 if i < skip_lines:
46 continue
47 if i == skip_lines and headers:
48 dset.headers = row
49 elif row:
50 if i > 0 and len(row) < dset.width:
51 row += [''] * (dset.width - len(row))
52 dset.append(row)
54 @classmethod
55 def detect(cls, stream, delimiter=None):
56 """Returns True if given stream is valid CSV."""
57 try:
58 csv.Sniffer().sniff(stream.read(2048), delimiters=delimiter or cls.DEFAULT_DELIMITER)
59 return True
60 except Exception:
61 return False