How to use the goodtables.processors.SchemaProcessor function in goodtables

To help you get started, we’ve selected a few goodtables examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github frictionlessdata / goodtables-py / tests / test_schema.py View on Github external
def test_standalone_row_limit_in_range(self):

        filepath = os.path.join(self.data_dir, 'row_limit_schema.csv')
        schema = os.path.join(self.data_dir, 'test_schema.json')
        with io.open(filepath) as stream:
            validator = processors.SchemaProcessor(row_limit=2, schema=schema)
            result, report, data = validator.run(stream)

            self.assertEqual(len(report.generate()['results']), 0)
github frictionlessdata / goodtables-py / tests / test_schema.py View on Github external
def test_standalone_schema_valid_simple(self):

        data_filepath = os.path.join(self.data_dir, 'contacts', 'people.csv')
        schema_filepath = os.path.join(self.data_dir, 'contacts',
                                       'schema_valid.json')
        with io.open(data_filepath) as data_stream, \
                 io.open(schema_filepath) as schema_stream:
            schema = json.load(schema_stream)
            validator = processors.SchemaProcessor(schema=schema)
            result, report, data = validator.run(data_stream)

            self.assertTrue(result)
github frictionlessdata / goodtables-py / tests / test_schema.py View on Github external
def test_standalone_process_extra_fields(self):
        filepath = os.path.join(self.data_dir, 'contacts', 'people.csv')
        schema = os.path.join(self.data_dir, 'contacts', 'schema_incomplete_fields.json')
        validator = processors.SchemaProcessor(schema=schema, case_insensitive_headers=True,
                                               process_extra_fields=True, result_level='info')
        result, report, data = validator.run(filepath)
        reports = report.generate()

        self.assertTrue(any('schema_008' in res.values() for res in reports['results']))
github frictionlessdata / goodtables-py / tests / test_schema.py View on Github external
def test_standalone_report_limit_out_range(self):

        limit = processors.SchemaProcessor.REPORT_LIMIT_MAX
        validator = processors.SchemaProcessor(report_limit=(limit + 1))

        self.assertEqual(validator.report_limit, limit)
github frictionlessdata / goodtables-py / tests / test_schema.py View on Github external
def test_standalone_fail_fast_true(self):

        filepath = os.path.join(self.data_dir, 'fail_fast_two_schema_errors.csv')
        schema = os.path.join(self.data_dir, 'test_schema.json')
        with io.open(filepath) as stream:
            validator = processors.SchemaProcessor(fail_fast=True, schema=schema)
            result, report, data = validator.run(stream)

            self.assertEqual(len(report.generate()['results']), 1)
github frictionlessdata / goodtables-py / tests / test_schema.py View on Github external
def test_standalone_case_insensitive_headers(self):
        filepath = os.path.join(self.data_dir, 'case_insensitive_headers.csv')
        schema = os.path.join(self.data_dir, 'test_schema.json')
        validator = processors.SchemaProcessor(schema=schema, case_insensitive_headers=True)
        result, report, data = validator.run(filepath)

        self.assertEqual(len(report.generate()['results']), 0)