def test_quoting(self):
class mydialect(csv.Dialect):
delimiter = ";"
escapechar = '\\'
doublequote = False
skipinitialspace = True
lineterminator = '\r\n'
quoting = csv.QUOTE_NONE
d = mydialect()
mydialect.quoting = None
self.assertRaises(csv.Error, mydialect)
mydialect.doublequote = True
mydialect.quoting = csv.QUOTE_ALL
mydialect.quotechar = '"'
d = mydialect()
mydialect.quotechar = "''"
self.assertRaises(csv.Error, mydialect)
mydialect.quotechar = 4
self.assertRaises(csv.Error, mydialect)
python类Dialect()的实例源码
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
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 __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def test_lineterminator(self):
class mydialect(csv.Dialect):
delimiter = ";"
escapechar = '\\'
doublequote = False
skipinitialspace = True
lineterminator = '\r\n'
quoting = csv.QUOTE_NONE
d = mydialect()
self.assertEqual(d.lineterminator, '\r\n')
mydialect.lineterminator = ":::"
d = mydialect()
self.assertEqual(d.lineterminator, ":::")
mydialect.lineterminator = 4
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"lineterminator" must be a string')
def test_quoting(self):
class mydialect(csv.Dialect):
delimiter = ";"
escapechar = '\\'
doublequote = False
skipinitialspace = True
lineterminator = '\r\n'
quoting = csv.QUOTE_NONE
d = mydialect()
self.assertEqual(d.quoting, csv.QUOTE_NONE)
mydialect.quoting = None
self.assertRaises(csv.Error, mydialect)
mydialect.doublequote = True
mydialect.quoting = csv.QUOTE_ALL
mydialect.quotechar = '"'
d = mydialect()
self.assertEqual(d.quoting, csv.QUOTE_ALL)
self.assertEqual(d.quotechar, '"')
self.assertTrue(d.doublequote)
mydialect.quotechar = "''"
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"quotechar" must be an 1-character string')
mydialect.quotechar = 4
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"quotechar" must be string, not int')
def test_lineterminator(self):
class mydialect(csv.Dialect):
delimiter = ";"
escapechar = '\\'
doublequote = False
skipinitialspace = True
lineterminator = '\r\n'
quoting = csv.QUOTE_NONE
d = mydialect()
self.assertEqual(d.lineterminator, '\r\n')
mydialect.lineterminator = ":::"
d = mydialect()
self.assertEqual(d.lineterminator, ":::")
mydialect.lineterminator = 4
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"lineterminator" must be a string')
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def format(self, value):
output = StringIO()
writer = csv.writer(output, List.Dialect)
writer.writerow(value)
value = output.getvalue()
return value[:-1]
def test_quoting(self):
class mydialect(csv.Dialect):
delimiter = ";"
escapechar = '\\'
doublequote = False
skipinitialspace = True
lineterminator = '\r\n'
quoting = csv.QUOTE_NONE
d = mydialect()
mydialect.quoting = None
self.assertRaises(csv.Error, mydialect)
mydialect.doublequote = True
mydialect.quoting = csv.QUOTE_ALL
mydialect.quotechar = '"'
d = mydialect()
mydialect.quotechar = "''"
self.assertRaises(csv.Error, mydialect)
mydialect.quotechar = 4
self.assertRaises(csv.Error, mydialect)
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def __init__(self,path='/home/data/corpora/FB15K/'):
self.path = path
# define tsv dialect
self.tsvDialect = csv.Dialect()
self.tsvDialect.delimiter = '\t'
# self.tsvDialect.delimiter = '"'
# Attributes Declaration
self.Ids = {}
self.Links = []
self.TestLinks = []
self.ValidLinks = []
self.RelationIds = {}
self.NodeTypes = {}
self.TypeIds = {}
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def format(self, value):
output = StringIO()
writer = csv.writer(output, List.Dialect)
writer.writerow(value)
value = output.getvalue()
return value[:-1]
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def format(self, value):
output = StringIO()
writer = csv.writer(output, List.Dialect)
writer.writerow(value)
value = output.getvalue()
return value[:-1]
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def format(self, value):
output = StringIO()
writer = csv.writer(output, List.Dialect)
writer.writerow(value)
value = output.getvalue()
return value[:-1]
def __init__(self, stream,
dialect: csv.Dialect, fields: list, converters: list):
reader = csv.reader(stream, dialect=dialect)
self.header = next(reader)
self.reader = reader
self.converters = converters
self.Record = namedtuple('Record', fields)
def __init__(self, stream, dialect: csv.Dialect, fields: list,
converters: list, field_index: list):
super().__init__(stream, dialect, fields, converters)
self.indexer = create_indexer(field_index)
def __call__(self, value):
if value is None or isinstance(value, list):
return value
try:
value = csv.reader([value], self.Dialect).next()
except csv.Error as error:
raise ValueError(error)
if self._validator is None:
return value
try:
for index, item in enumerate(value):
value[index] = self._validator(item)
except ValueError as error:
raise ValueError('Could not convert item {}: {}'.format(index, error))
return value
def format(self, value):
output = StringIO()
writer = csv.writer(output, List.Dialect)
writer.writerow(value)
value = output.getvalue()
return value[:-1]
def test_incomplete_dialect(self):
class myexceltsv(csv.Dialect):
delimiter = "\t"
self.assertRaises(csv.Error, myexceltsv)
def test_quoting(self):
class mydialect(csv.Dialect):
delimiter = ";"
escapechar = '\\'
doublequote = False
skipinitialspace = True
lineterminator = '\r\n'
quoting = csv.QUOTE_NONE
d = mydialect()
self.assertEqual(d.quoting, csv.QUOTE_NONE)
mydialect.quoting = None
self.assertRaises(csv.Error, mydialect)
mydialect.doublequote = True
mydialect.quoting = csv.QUOTE_ALL
mydialect.quotechar = '"'
d = mydialect()
self.assertEqual(d.quoting, csv.QUOTE_ALL)
self.assertEqual(d.quotechar, '"')
self.assertTrue(d.doublequote)
mydialect.quotechar = "''"
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"quotechar" must be a 1-character string')
mydialect.quotechar = 4
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"quotechar" must be string, not int')
def test_delimiter(self):
class mydialect(csv.Dialect):
delimiter = ";"
escapechar = '\\'
doublequote = False
skipinitialspace = True
lineterminator = '\r\n'
quoting = csv.QUOTE_NONE
d = mydialect()
self.assertEqual(d.delimiter, ";")
mydialect.delimiter = ":::"
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"delimiter" must be a 1-character string')
mydialect.delimiter = ""
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"delimiter" must be a 1-character string')
mydialect.delimiter = u","
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"delimiter" must be string, not unicode')
mydialect.delimiter = 4
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"delimiter" must be string, not int')
def test_lineterminator(self):
class mydialect(csv.Dialect):
delimiter = ";"
escapechar = '\\'
doublequote = False
skipinitialspace = True
lineterminator = '\r\n'
quoting = csv.QUOTE_NONE
d = mydialect()
self.assertEqual(d.lineterminator, '\r\n')
mydialect.lineterminator = ":::"
d = mydialect()
self.assertEqual(d.lineterminator, ":::")
mydialect.lineterminator = 4
with self.assertRaises(csv.Error) as cm:
mydialect()
self.assertEqual(str(cm.exception),
'"lineterminator" must be a string')
def test_incomplete_dialect(self):
class myexceltsv(csv.Dialect):
delimiter = "\t"
self.assertRaises(csv.Error, myexceltsv)
