def generate_schema()

in jobs/fxci-taskcluster-export/fxci_etl/schemas.py [0:0]

• 25 lines of code
• 8 McCabe index (conditional complexity)

def generate_schema(cls):
    assert is_dataclass(cls)
    schema = []
    for field in fields(cls):
        _type = field.type
        origin = get_origin(_type)
        args = get_args(_type)

        if origin is Union and type(None) in args:
            mode = "NULLABLE"
            _type = [arg for arg in args if arg is not type(None)][0]
            origin = get_origin(_type)
            args = get_args(_type)
        else:
            mode = "REQUIRED"

        if origin is list:
            mode = "REPEATED"
            _type = args[0]

        if is_dataclass(_type):
            nested_schema = generate_schema(_type)
            schema.append(
                SchemaField(field.name, "RECORD", mode=mode, fields=nested_schema)
            )
        else:
            schema.append(SchemaField(field.name, _type.name, mode=mode))
    return schema