Python script to validate a JSON file against a schema

json_schema_validator.py

# -*- coding: utf-8 -*-
""" Purpose: Use Python schema module to validate JSON files vs a schema.

Created: 2017-09-05
Author:  [email protected]

Usage::

    python json_schema_validator.py --json=ajsonfile.json --schema=aschemafile.json
"""
from __future__ import print_function
# lib
import argparse
import json
import logging
import sys
# third party
import jsonschema
from jsonschema.exceptions import SchemaError

log = logging.getLogger(__name__)
LOG_LEVEL = 'INFO'


class JSONValidationError(Exception):
    """Error handling for JSONValidator."""

    pass


class JSONValidator(object):
    '''
    Enables validation of a schema, as well as a JSON file against that schema.
    '''

    def __init__(self, **kwargs):
        self.json_file = kwargs.get('json_file')
        self.json_schema = kwargs.get('json_schema')
        self.schema_string = self.validate_schema(self.json_schema )

    def validate_schema(self, schema_filename, version=4):
        """Test that a schema used to validate data is itself valid.

        Return the validated schema as a JSON string.
        """
        if version == 3:
            from jsonschema import Draft3Validator as DraftValidator
        elif version == 4:
            from jsonschema import Draft4Validator as DraftValidator
        else:
            raise SOSServiceError(
                'Invalid schema validator number:"%s"' % version)
        the_schema = None
        try:
            with open(schema_filename, 'r') as data_file:
                the_schema = json.load(data_file)
            DraftValidator.check_schema(the_schema)
            result = True
        except (ValueError, IOError) as err:
            log.error('Unable to process file: "%s" (%s)',
                      schema_filename, err)
            result = False
        except SchemaError as err:
            log.error('Unable to validate schema: "%s" (%s)',
                      schema_filename, err)
            result = False
        if not result:
            raise JSONValidationError(
                'Unable to complete validation of *schema*: %s' % schema_filename)
        return the_schema

    def validate(self):
        """"Validate against the specified schema."""
        try:
            with open(self.json_file, 'r') as data_file:
                self.json_string = json.load(data_file)
            jsonschema.validate(self.json_string, self.schema_string)
            return True
        except SchemaError as err:
            log.error('Unable to validate against schema: "%s" (%s)',
                      self.json_schema, err)
        except (ValueError, IOError) as err:
            log.error('Unable to process the JSON file: "%s" (%s)',
                      self.json_file, err)


def main(args):
    """Process args."""
    logging.basicConfig(
        stream=sys.stdout,
        format='%(asctime)s %(levelname)-8s [%(filename)s:%(lineno)d %(funcName)s()] %(message)s',
        level=getattr(logging, args.loglevel))
    # do check
    try:
        JSONValidator(json_file=args.json, json_schema=args.schema).validate()
        print('Validated!')
    except Exception as err:
        print('NOT validated!')
        print(err)


if __name__ == '__main__':
    PARSER = argparse.ArgumentParser(
        description="Perform validation of JSON file vs a supplied schema file"
    )
    PARSER.add_argument(
        '-j', '--json',
        help="Name of JSON file to validate")
    PARSER.add_argument(
        '-s', '--schema',
        help="Name of JSON schema file to use for validation")
    PARSER.add_argument(
        '-ll', '--loglevel', default=LOG_LEVEL,
        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
        help="Set log level for service (%s)" % LOG_LEVEL)
    ARGS = PARSER.parse_args()
    LOG_LEVEL = ARGS.loglevel
    main(ARGS)