Creates documents for elasticsearch_dsl

elasticsearch_index_to_dsl_document.py

from inspect import isclass
import json
import sys

import elasticsearch_dsl.field


def snake_to_camel(word):
    return ''.join(x.capitalize() or '_' for x in word.split('_'))


field_types = (
        field_type
        for typename in dir(elasticsearch_dsl.field)
        for field_type in (getattr(elasticsearch_dsl.field, typename),)
        if isclass(field_type) and issubclass(field_type, elasticsearch_dsl.field.Field)
)

field_types_by_name = {
        field_type.name: field_type
        for field_type in field_types
}


def convert_mapping(class_name, mapping, used_type_names, converted_mappings, parent="Document", index_name=None):
    mappings_to_convert = []
    properties = mapping["properties"]
    class_body_lines = []
    for property_name in properties:
        field_mapping = properties[property_name]
        field_type_name = field_mapping.get("type", "object")
        field_type = field_types_by_name[field_type_name]
        field_type_class_name = field_type.__name__
        if "properties" in field_mapping and field_type_name in ("object", "nested"):
            object_class_name = f"{class_name}{snake_to_camel(property_name)}"
            class_body_lines.append(f"  {property_name} = {field_type_class_name}({object_class_name})")
            mappings_to_convert.append({
                "class_name": object_class_name,
                "mapping": field_mapping
            })
        else:
            type_kwargs = ", ".join(f"{item[0]}=\"{item[1]}\"" for item in field_mapping.items() if item[0] in ["analyzer"])
            class_body_lines.append(f"""  {property_name} = {field_type_class_name}({type_kwargs})""")
        used_type_names.add(field_type_class_name)
    class_body_lines.sort()
    if index_name:
        class_body_lines = [
            "  class Index:",
            f"    name = \"{index_name}\"",
            *class_body_lines
        ]
    class_body = f"class {class_name}({parent}):\n" + "\n".join(class_body_lines)
    for config in mappings_to_convert:
        convert_mapping(config["class_name"], config["mapping"], used_type_names, converted_mappings, parent="InnerDoc")
    converted_mappings[class_name] = class_body


if __name__ == "__main__":
    # Get this from http://elasticsearch:9200/*
    indexes = json.load(sys.stdin)
    used_type_names = set()
    converted_mappings = {}
    for index in indexes:
        class_name = snake_to_camel(index)
        convert_mapping(class_name, indexes[index]["mappings"], used_type_names, converted_mappings, index_name=index)
    imports = ", ".join(sorted(used_type_names))
    print("from elasticsearch_dsl import Document, InnerDoc")
    print(f"from elasticsearch_dsl.field import {imports}")
    for mapping_name in converted_mappings:
        print()
        print()
        print(converted_mappings[mapping_name])