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, [])
python类list_dialects()的实例源码
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 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_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_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_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_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_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_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_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 main():
try:
import setproctitle
setproctitle.setproctitle(name)
except ImportError:
pass
dialects = csv.list_dialects()
dialects.sort()
dialects.insert(0, 'sniff')
# CLI arguments
parser = argparse.ArgumentParser(prog=name, description=description)
parser.add_argument('-V', '--version', action='version', version="%(prog)s " + version)
parser.add_argument('-C', '--config', help='Use a different configuration file')
parser.add_argument('-s', '--section', help='Configuration file section', default='DEFAULT')
parser.add_argument('-c', '--channel', help='Send to this channel or @username')
parser.add_argument('-U', '--url', help='Mattermost webhook URL')
parser.add_argument('-u', '--username', help='Username')
parser.add_argument('-i', '--icon', help='Icon')
group = parser.add_mutually_exclusive_group()
group.add_argument('-t', '--tabular', metavar='DIALECT', const='sniff',
nargs='?', choices=dialects,
help='Parse input as CSV and format it as a table (DIALECT can be one of %(choices)s)')
group.add_argument('-y', '--syntax', default='auto')
parser.add_argument('-I', '--info', action='store_true',
help='Include file information in message')
parser.add_argument('-n', '--dry-run', '--just-print', action='store_true',
help="Don't send, just print the payload")
parser.add_argument('-f', '--file', default='-',
help="Read content from FILE. If - reads from standard input (DEFAULT: %(default)s)")
args = parser.parse_args()
if args.file == '-':
message = sys.stdin.read()
filename = None
else:
message = ''
filename = args.file
try:
payload = send(args.channel, message, filename, args.url,
args.username, args.icon, args.syntax, args.tabular,
args.info, args.dry_run, args.section, name,
args.config)
except (configparser.Error, TypeError, RuntimeError) as e:
sys.exit(str(e))
if args.dry_run:
print(payload)
def _get_child_types(self):
_result = {}
_result["datatypes"] = {"type": "string", "pattern": "(integer|string|string(\(.*\))|serial|timestamp)"}
_result["db_types"] = {"type": "string", "enum": db_types()}
_result["and_or"] = {"type": "string", "enum": and_or()}
_result["in_types"] = {"type": "string", "enum": in_types()}
_result["index_types"] = {"type": "string", "enum": index_types()}
_result["constraint_types"] = {"type": "string", "enum": constraint_types()}
_result["set_operator"] = {"type": "string", "enum": set_operator()}
_result["quoting"] = {"type": "string", "enum": quoting_types()}
_result["csv_dialects"] = {"type": "string", "enum": list_dialects()}
_result["join_types"] = {"type": "string", "enum": join_types()}
_result["xpath_data_format"] = {"type": "string", "enum": xpath_data_formats()}
def make_one_of(_classes):
return {"type": "object",
"anyOf":
[{"properties": {x: {"$ref": "#/definitions/" + x}}} for x in _classes]
}
_result["statement"] = make_one_of(verbs())
_result["condition_part"] = make_one_of(condition_part())
_result["TabularExpressionItem"] = make_one_of(tabular_expression_item_types())
_result["data_source_types"] = make_one_of(data_source_types())
_result["ArrayString"] = {"type": "array", "items": {"type": "string"}}
_result["ArrayList"] = {"type": "array"}
_result["ArrayParameterString"] = self._child_array_of(['#/definitions/ParameterString'])
_result["ArrayParameterConstraint"] = self._child_array_of(['#/definitions/ParameterConstraint'])
_result["ArrayParameterColumndefinition"] = self._child_array_of(['#/definitions/ParameterColumndefinition'])
_result["ArrayParameterSource"] = self._child_array_of(['#/definitions/ParameterSource'])
_result["ArrayParameterWhen"] = self._child_array_of(['#/definitions/ParameterWhen'])
_result["ArrayParameterIdentifier"] = self._child_array_of(['#/definitions/ParameterIdentifier'])
_result["ArrayStatement"] = self._child_array_of(['#/definitions/statement'])
_result["ArrayParameterOrderByItem"] = self._child_array_of(['#/definitions/ParameterOrderByItem'])
_result["ArrayParameterCondition"] = self._child_array_of(['#/definitions/ParameterCondition'])
_result["ArrayParameterField"] = self._child_array_of(['#/definitions/ParameterField'])
_result["ArrayParameterAssignment"] = self._child_array_of(['#/definitions/ParameterAssignment'])
_result["ArrayExpressionItem"] = self._child_array_of(make_one_of(expression_item_types()))
_result["ArrayTabularExpressionItem"] = self._child_array_of(make_one_of(tabular_expression_item_types()))
return _result
parsers.py 文件源码
项目：PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 33
收藏 0
点赞 0
评论 0
def __init__(self, f, engine=None, **kwds):
self.f = f
if engine is not None:
engine_specified = True
else:
engine = 'python'
engine_specified = False
self._engine_specified = kwds.get('engine_specified', engine_specified)
if kwds.get('dialect') is not None:
dialect = kwds['dialect']
if dialect in csv.list_dialects():
dialect = csv.get_dialect(dialect)
kwds['delimiter'] = dialect.delimiter
kwds['doublequote'] = dialect.doublequote
kwds['escapechar'] = dialect.escapechar
kwds['skipinitialspace'] = dialect.skipinitialspace
kwds['quotechar'] = dialect.quotechar
kwds['quoting'] = dialect.quoting
if kwds.get('header', 'infer') == 'infer':
kwds['header'] = 0 if kwds.get('names') is None else None
self.orig_options = kwds
# miscellanea
self.engine = engine
self._engine = None
options = self._get_options_with_defaults(engine)
self.chunksize = options.pop('chunksize', None)
self.squeeze = options.pop('squeeze', False)
# might mutate self.engine
self.options, self.engine = self._clean_options(options, engine)
if 'has_index_names' in kwds:
self.options['has_index_names'] = kwds['has_index_names']
self._make_engine(self.engine)
