def __call__(self, context):
if isinstance(self._object, types.StringTypes):
self._object = context[self._object]
# get dialect object
if isinstance(self._dialect, types.StringTypes):
dialect = csv.get_dialect(self._dialect)
if self._path.startswith("memory:"):
buffer_ = StringIO.StringIO()
self._write_object(buffer_, dialect)
buffer_.seek(0)
context[self._path[len("memory:"):]] = buffer_
else:
with open(self._path, "w") as f:
self._write_object(f, dialect)
return self._path
python类get_dialect()的实例源码
def test_registry_badargs(self):
self.assertRaises(TypeError, csv.list_dialects, None)
self.assertRaises(TypeError, csv.get_dialect)
self.assertRaises(csv.Error, csv.get_dialect, None)
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
self.assertRaises(TypeError, csv.unregister_dialect)
self.assertRaises(csv.Error, csv.unregister_dialect, None)
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
self.assertRaises(TypeError, csv.register_dialect, None)
self.assertRaises(TypeError, csv.register_dialect, None, None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
badargument=None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
quoting=None)
self.assertRaises(TypeError, csv.register_dialect, [])
def test_registry_badargs(self):
self.assertRaises(TypeError, csv.list_dialects, None)
self.assertRaises(TypeError, csv.get_dialect)
self.assertRaises(csv.Error, csv.get_dialect, None)
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
self.assertRaises(TypeError, csv.unregister_dialect)
self.assertRaises(csv.Error, csv.unregister_dialect, None)
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
self.assertRaises(TypeError, csv.register_dialect, None)
self.assertRaises(TypeError, csv.register_dialect, None, None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
badargument=None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
quoting=None)
self.assertRaises(TypeError, csv.register_dialect, [])
def test_bad_dialect(self):
# Unknown parameter
self.assertRaises(TypeError, csv.reader, [], bad_attr = 0)
# Bad values
self.assertRaises(TypeError, csv.reader, [], delimiter = None)
self.assertRaises(TypeError, csv.reader, [], quoting = -1)
self.assertRaises(TypeError, csv.reader, [], quoting = 100)
# See issue #22995
## def test_copy(self):
## for name in csv.list_dialects():
## dialect = csv.get_dialect(name)
## self.assertRaises(TypeError, copy.copy, dialect)
## def test_pickle(self):
## for name in csv.list_dialects():
## dialect = csv.get_dialect(name)
## for proto in range(pickle.HIGHEST_PROTOCOL + 1):
## self.assertRaises(TypeError, pickle.dumps, dialect, proto)
def test_registry_badargs(self):
self.assertRaises(TypeError, csv.list_dialects, None)
self.assertRaises(TypeError, csv.get_dialect)
self.assertRaises(csv.Error, csv.get_dialect, None)
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
self.assertRaises(TypeError, csv.unregister_dialect)
self.assertRaises(csv.Error, csv.unregister_dialect, None)
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
self.assertRaises(TypeError, csv.register_dialect, None)
self.assertRaises(TypeError, csv.register_dialect, None, None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
badargument=None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
quoting=None)
self.assertRaises(TypeError, csv.register_dialect, [])
def test_bad_dialect(self):
# Unknown parameter
self.assertRaises(TypeError, csv.reader, [], bad_attr = 0)
# Bad values
self.assertRaises(TypeError, csv.reader, [], delimiter = None)
self.assertRaises(TypeError, csv.reader, [], quoting = -1)
self.assertRaises(TypeError, csv.reader, [], quoting = 100)
# See issue #22995
## def test_copy(self):
## for name in csv.list_dialects():
## dialect = csv.get_dialect(name)
## self.assertRaises(TypeError, copy.copy, dialect)
## def test_pickle(self):
## for name in csv.list_dialects():
## dialect = csv.get_dialect(name)
## for proto in range(pickle.HIGHEST_PROTOCOL + 1):
## self.assertRaises(TypeError, pickle.dumps, dialect, proto)
def test_registry_badargs(self):
self.assertRaises(TypeError, csv.list_dialects, None)
self.assertRaises(TypeError, csv.get_dialect)
self.assertRaises(csv.Error, csv.get_dialect, None)
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
self.assertRaises(TypeError, csv.unregister_dialect)
self.assertRaises(csv.Error, csv.unregister_dialect, None)
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
self.assertRaises(TypeError, csv.register_dialect, None)
self.assertRaises(TypeError, csv.register_dialect, None, None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
badargument=None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
quoting=None)
self.assertRaises(TypeError, csv.register_dialect, [])
def test_registry_badargs(self):
self.assertRaises(TypeError, csv.list_dialects, None)
self.assertRaises(TypeError, csv.get_dialect)
self.assertRaises(csv.Error, csv.get_dialect, None)
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
self.assertRaises(TypeError, csv.unregister_dialect)
self.assertRaises(csv.Error, csv.unregister_dialect, None)
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
self.assertRaises(TypeError, csv.register_dialect, None)
self.assertRaises(TypeError, csv.register_dialect, None, None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
badargument=None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
quoting=None)
self.assertRaises(TypeError, csv.register_dialect, [])
def test_registry_badargs(self):
self.assertRaises(TypeError, csv.list_dialects, None)
self.assertRaises(TypeError, csv.get_dialect)
self.assertRaises(csv.Error, csv.get_dialect, None)
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
self.assertRaises(TypeError, csv.unregister_dialect)
self.assertRaises(csv.Error, csv.unregister_dialect, None)
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
self.assertRaises(TypeError, csv.register_dialect, None)
self.assertRaises(TypeError, csv.register_dialect, None, None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
badargument=None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
quoting=None)
self.assertRaises(TypeError, csv.register_dialect, [])
def test_registry_badargs(self):
self.assertRaises(TypeError, csv.list_dialects, None)
self.assertRaises(TypeError, csv.get_dialect)
self.assertRaises(csv.Error, csv.get_dialect, None)
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
self.assertRaises(TypeError, csv.unregister_dialect)
self.assertRaises(csv.Error, csv.unregister_dialect, None)
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
self.assertRaises(TypeError, csv.register_dialect, None)
self.assertRaises(TypeError, csv.register_dialect, None, None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
badargument=None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
quoting=None)
self.assertRaises(TypeError, csv.register_dialect, [])
def test_registry_badargs(self):
self.assertRaises(TypeError, csv.list_dialects, None)
self.assertRaises(TypeError, csv.get_dialect)
self.assertRaises(csv.Error, csv.get_dialect, None)
self.assertRaises(csv.Error, csv.get_dialect, "nonesuch")
self.assertRaises(TypeError, csv.unregister_dialect)
self.assertRaises(csv.Error, csv.unregister_dialect, None)
self.assertRaises(csv.Error, csv.unregister_dialect, "nonesuch")
self.assertRaises(TypeError, csv.register_dialect, None)
self.assertRaises(TypeError, csv.register_dialect, None, None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch", 0, 0)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
badargument=None)
self.assertRaises(TypeError, csv.register_dialect, "nonesuch",
quoting=None)
self.assertRaises(TypeError, csv.register_dialect, [])
def __init__(self, n_samples, out_file, pid, lid, keep_cols,
n_retry, delimiter, dataset, pred_name, ui, file_context,
fast_mode, encoding, skip_row_id, output_delimiter):
self.n_samples = n_samples
self.out_file = out_file
self.project_id = pid
self.model_id = lid
self.keep_cols = keep_cols
self.n_retry = n_retry
self.delimiter = delimiter
self.dataset = dataset
self.pred_name = pred_name
self.out_stream = None
self._ui = ui
self.file_context = file_context
self.fast_mode = fast_mode
self.encoding = encoding
self.skip_row_id = skip_row_id
self.output_delimiter = output_delimiter
# dataset_dialect and writer_dialect are set by
# investigate_encoding_and_dialect in utils
self.dialect = csv.get_dialect('dataset_dialect')
self.writer_dialect = csv.get_dialect('writer_dialect')
self.scoring_succeeded = False # Removes shelves when True
self.is_open = False # Removes shelves when True
def close(self):
if not self.is_open:
self._ui.debug('CLOSE CALLED ON CLOSED RUNCONTEXT')
return
self.is_open = False
self._ui.debug('CLOSE CALLED ON RUNCONTEXT')
self.dialect = csv.get_dialect('dataset_dialect')
self.writer_dialect = csv.get_dialect('writer_dialect')
values = ['delimiter', 'doublequote', 'escapechar', 'lineterminator',
'quotechar', 'quoting', 'skipinitialspace', 'strict']
self.dialect = {k: getattr(self.dialect, k) for k in values if
hasattr(self.dialect, k)}
self.writer_dialect = {k: getattr(self.writer_dialect, k) for k
in values if hasattr(self.writer_dialect, k)}
self.db.sync()
self.db.close()
if self.out_stream is not None:
self.out_stream.close()
def __call__(self, context):
# get Dialect object
if isinstance(self._dialect, types.StringTypes):
dialect = csv.get_dialect(self._dialect)
# get content
content = self._content
if self._content is None:
if self._path.startswith(HTTP_SCHEMA):
content = requests.get(self._path).text
else:
with open(self._path, "r") as f:
content = f.read()
if isinstance(content, types.StringTypes):
content = StringIO.StringIO(content)
result = []
csv_reader = csv.reader(content, dialect=dialect)
for row in csv_reader:
self._handle_record(row, result.append)
return self._handle_result(context, result)
def test_registry(self):
class myexceltsv(csv.excel):
delimiter = "\t"
name = "myexceltsv"
expected_dialects = csv.list_dialects() + [name]
expected_dialects.sort()
csv.register_dialect(name, myexceltsv)
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, '\t')
got_dialects = sorted(csv.list_dialects())
self.assertEqual(expected_dialects, got_dialects)
def test_register_kwargs(self):
name = 'fedcba'
csv.register_dialect(name, delimiter=';')
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, ';')
self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
def test_registry(self):
class myexceltsv(csv.excel):
delimiter = "\t"
name = "myexceltsv"
expected_dialects = csv.list_dialects() + [name]
expected_dialects.sort()
csv.register_dialect(name, myexceltsv)
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, '\t')
got_dialects = sorted(csv.list_dialects())
self.assertEqual(expected_dialects, got_dialects)
def test_register_kwargs(self):
name = 'fedcba'
csv.register_dialect(name, delimiter=';')
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, ';')
self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
def test_registry(self):
class myexceltsv(csv.excel):
delimiter = "\t"
name = "myexceltsv"
expected_dialects = csv.list_dialects() + [name]
expected_dialects.sort()
csv.register_dialect(name, myexceltsv)
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, '\t')
got_dialects = sorted(csv.list_dialects())
self.assertEqual(expected_dialects, got_dialects)
def test_register_kwargs(self):
name = 'fedcba'
csv.register_dialect(name, delimiter=';')
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, ';')
self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
def test_registry(self):
class myexceltsv(csv.excel):
delimiter = "\t"
name = "myexceltsv"
expected_dialects = csv.list_dialects() + [name]
expected_dialects.sort()
csv.register_dialect(name, myexceltsv)
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, '\t')
got_dialects = sorted(csv.list_dialects())
self.assertEqual(expected_dialects, got_dialects)
def test_register_kwargs(self):
name = 'fedcba'
csv.register_dialect(name, delimiter=';')
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, ';')
self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
def test_registry(self):
class myexceltsv(csv.excel):
delimiter = "\t"
name = "myexceltsv"
expected_dialects = csv.list_dialects() + [name]
expected_dialects.sort()
csv.register_dialect(name, myexceltsv)
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, '\t')
got_dialects = sorted(csv.list_dialects())
self.assertEqual(expected_dialects, got_dialects)
def test_register_kwargs(self):
name = 'fedcba'
csv.register_dialect(name, delimiter=';')
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, ';')
self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
def test_registry(self):
class myexceltsv(csv.excel):
delimiter = "\t"
name = "myexceltsv"
expected_dialects = csv.list_dialects() + [name]
expected_dialects.sort()
csv.register_dialect(name, myexceltsv)
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, '\t')
got_dialects = sorted(csv.list_dialects())
self.assertEqual(expected_dialects, got_dialects)
def test_register_kwargs(self):
name = 'fedcba'
csv.register_dialect(name, delimiter=';')
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, ';')
self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
def test_registry(self):
class myexceltsv(csv.excel):
delimiter = "\t"
name = "myexceltsv"
expected_dialects = csv.list_dialects() + [name]
expected_dialects.sort()
csv.register_dialect(name, myexceltsv)
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, '\t')
got_dialects = sorted(csv.list_dialects())
self.assertEqual(expected_dialects, got_dialects)
def test_register_kwargs(self):
name = 'fedcba'
csv.register_dialect(name, delimiter=';')
self.addCleanup(csv.unregister_dialect, name)
self.assertEqual(csv.get_dialect(name).delimiter, ';')
self.assertEqual([['X', 'Y', 'Z']], list(csv.reader(['X;Y;Z'], name)))
def __init__(self, filename, renmwo = True):
self.filename = filename
self.renmwo = "%s.~renmwo%d~" % (filename, os.getpid()) if renmwo else filename
self.file = open(self.renmwo, "w")
self.writer = csv.writer(self.file, dialect = csv.get_dialect("excel-tab"))
def __init__(self, filename, renmwo = True):
self.filename = filename
self.renmwo = "%s.~renmwo%d~" % (filename, os.getpid()) if renmwo else filename
self.file = open(self.renmwo, "w")
self.writer = csv.writer(self.file, dialect = csv.get_dialect("excel-tab"))
