def load_schemas()

in src/loading_manifest/csv_to_json.py [0:0]

• 53 lines of code
• 15 McCabe index (conditional complexity)

def load_schemas(schema_path, schema_ns_name=None, schema_ns_value=None):
    def list_schema_files(path, file_list):
        files = os.listdir(path)
        for file in files:
            full_path = os.path.join(path, file)
            if os.path.isfile(full_path):
                if file.endswith('.json'):
                    file_list.append(full_path)
            elif os.path.isdir(full_path):
                list_schema_files(full_path, file_list)

    dict_schemas = dict()

    # load all json files
    file_list = []
    list_schema_files(schema_path, file_list)
    for schema_file in file_list:
        with open(schema_file, 'r', encoding='utf-8') as fp:
            a_schema = json.load(fp)
            if schema_ns_name is not None and len(schema_ns_name) > 0:
                a_schema = replace_json_namespace(a_schema, schema_ns_name+":", schema_ns_value+":")
            id = a_schema.get('$id')
            if id is None:
                id = a_schema.get('$ID')
            if id is not None:
                dict_schemas[id] = a_schema

            # for top level resource,
            if a_schema.get('properties', {}).get('ResourceHomeRegionID') is not None:
                # remove 'required"
                if a_schema.get('required') is not None:
                    del a_schema['required']
                # remove 'additionalProperties"
                if a_schema.get('additionalProperties') is not None:
                    del a_schema['additionalProperties']
                # remove resource/type id version
                if a_schema.get('properties', {}).get('ResourceTypeID', {}).get('pattern') is not None:
                    a_schema['properties']['ResourceTypeID']['pattern'] = \
                        a_schema['properties']['ResourceTypeID']['pattern'].replace(':[0-9]+', ':[0-9]*')
                if a_schema.get('properties', {}).get('ResourceID', {}).get('pattern') is not None:
                    a_schema['properties']['ResourceID']['pattern'] = \
                        a_schema['properties']['ResourceID']['pattern'].replace(':[0-9]+', ':[0-9]*')

    # resolve latest version
    dict_latest_key = dict()
    dict_latest_version = dict()
    for key, val in dict_schemas.items():
        # strip the version at the end
        key_parts = key.split('/')
        key_version = None
        if len(key_parts) > 1:
            try:
                key_version = int(key_parts[-1])
            except ValueError:
                pass
        if key_version is not None:
            key_latest_id = '/'.join(key_parts[:-1]) + '/'
            previous_key_version = dict_latest_version.get(key_latest_id, None)
            if previous_key_version is None or key_version > previous_key_version:
                dict_latest_version[key_latest_id] = key_version
                dict_latest_key[key_latest_id] = key
    for latest_key, key in dict_latest_key.items():
        dict_schemas[latest_key] = dict_schemas[key]

    return dict_schemas