detect_avro_schema_cycle.py

# Copyright 2021 Aiven Oy https://aiven.io
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


def schema(s, seen):
    if isinstance(s, str):
        assert s not in seen, f"found cycle with {s}"
    elif isinstance(s, list):
        union(s, seen)
    elif isinstance(s, dict):
        complex_(s, seen)


def field(f, seen):
    schema(f["type"], seen)


def record(r, seen):
    seen = set(seen)
    assert r["type"] == "record", f"expected record got {r['type']}"
    seen.add(r["name"])
    for f in r["fields"]:
        field(f, seen)


def array(a, seen):
    assert a["type"] == "array", f"expected array got {a['type']}"
    schema(a["items"], seen)


def map(m, seen):
    assert m["type"] == "map", f"expected map got {m['type']}"
    schema(m["values"], seen)


def union(u, seen):
    for x in u:
        schema(x, seen)


def enum(e, seen):
    pass


def string(e, seen):
    pass


def int(e, seen):
    pass


def bytes(e, seen):
    pass


def complex_(c, seen):
    globals()[c["type"]](c, seen)


if __name__ == "__main__":
    import argparse, json
    parser = argparse.ArgumentParser()
    parser.add_argument('schema')
    args = parser.parse_args()
    schema(json.load(open(args.schema)), set())