Generate openapi swagger object definition form avro schema

avro_schema_to_swagger.py

"""
Convert Avro schema files to swagger object definitions.
"""
import argparse
import json
from dataclasses import dataclass
from typing import Dict, List

parser = argparse.ArgumentParser(
    description="Convert avro schema file to swagger object definition"
)
parser.add_argument(
    "file-path",
    metavar="File Path",
    type=str,
    nargs="+",
    help="List of files to convert",
)


@dataclass
class AvroField:
    name: str
    typ: str


def extract_avro_field(avro_field_raw: Dict) -> AvroField:
    return AvroField(name=avro_field_raw["name"], typ=avro_field_raw["type"])


def generate_required_list(fields: List[AvroField]) -> str:
    required = "required:\n"
    for field in fields:
        required += "  "
        required += f"- {field.name}"
        required += "\n"
    return required


def generate_properties(fields: List[AvroField]) -> str:
    properties = "properties:\n"
    for field in fields:
        properties += "  "
        properties += f"{field.name}:\n"
        properties += "    "
        properties += f'type: "{field.typ}"\n'
    return properties


def generate_swagger_definition(avro_schema: Dict) -> str:
    fields = list(map(extract_avro_field, avro_schema["fields"]))
    required_list = generate_required_list(fields)
    properties = generate_properties(fields)
    return required_list + properties


def main():
    args = vars(parser.parse_args())
    files = args["file-path"]
    for avro_files in files:
        with open(avro_files, "r", encoding="utf-8") as avro_schema_file:
            avro_schema = json.load(avro_schema_file)
            swagger_def = generate_swagger_definition(avro_schema)
            print(swagger_def)


main()