Python script to generate Basic MongoDB Schema Validation. Doesn't support enums and other datatypes yet.

mongodb_schema_validation_creator_basic.py

import json
import datetime
from bson import ObjectId


def get_mongodb_type(data_type):
    if data_type is None:
        return 'null'
    elif type(data_type) == type(""):
        return "string"
    elif type(data_type) == type(1):
        return 'int'
    elif type(data_type) == type({'': ''}):
        return 'object'
    elif type(data_type) == type([]):
        return 'array'
    elif type(data_type) == type(datetime.datetime.now()):
        return 'date'
    elif type(data_type) == type(1.0):
        return 'double'
    elif type(data_type) == type(True):
        return 'bool'
    elif type(data_type) == type(ObjectId()):
        return 'objectId'
    elif type(data_type) == type(b""):
        return 'binData'
    elif data_type == 'timestamp':
        return 'timestamp'


def get_mongodb_description(bson_type, required=False):
    if bson_type == 'string':
        string = "must be a string"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'int':
        string = "must be an integer"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'double':
        string = "must be a double/float"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'array':
        string = "must be an array"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'date':
        string = "must be a datetime object"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'object':
        string = "must be an object"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'objectId':
        string = "must be a bson ObjectId"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'binData':
        string = "must be a binary string"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'timestamp':
        string = "must be a timestamp"
        string += " and is required" if required else ''
        return string
    elif bson_type == 'bool':
        string = "must be a boolean"
        string += " and is required" if required else ''
        return string
    else:
        string = "can be null"
        return string


def create_schema_validator(fields_and_types: list, required_fields: list):
    validator = {'validator':
        {'$jsonSchema': {
            'bsonType': "object",
            'required': required_fields,
            'properties': {}
        }},
        'validationLevel': "moderate",
        'validationAction': "warn"
    }
    properties = {}
    for value in fields_and_types:
        bson_type = get_mongodb_type(value[1])
        description = get_mongodb_description(bson_type=bson_type,
                                              required=True if value[0] in required_fields else False)
        property_object = {'bsonType': bson_type, 'description': description}
        properties[value[0]] = property_object
    validator['validator']['$jsonSchema']['properties'] = properties
    return validator

# Sample Data
collection_name = 'users'
user_collection_data = {
    'fields_and_types': [('user_id', ObjectId()),
                         ('organisation_id', ""),
                         ('organisation_name', ""),
                         ('email', ""),
                         ('password', b""),
                         ('first_name', ""),
                         ('last_name', ""),
                         ('country', ""),
                         ('phone_number', ""),
                         ('role', []),
                         ('session_token', ""),
                         ('is_deleted', True),
                         ],
    'required_fields': ['user_id', 'organisation_id', 'organisation_name', 'email', 'password',
                        'first_name', 'last_name', 'country', 'phone_number', 'role',
                        'session_token']
}

required_fields = []
for value in collection_data['fields_and_types']:
    required_fields.append(value[0])
print(required_fields)

schema = create_schema_validator(fields_and_types=user_collection_data['fields_and_types'],
                                 required_fields=required_fields)
schema = json.dumps(schema, sort_keys=True, indent=4) # JSON safe for MongoDB Shell.

# Executable MongoDB Shell Script:
print("db.createCollection(\"" + collection_name + "\", " + schema + ")")