def _iter_response_scopes_by_schema_base()

in src/aaz_dev/cli/controller/az_operation_generator.py [0:0]

• 67 lines of code
• 29 McCabe index (conditional complexity)

def _iter_response_scopes_by_schema_base(schema, name, scope_define, cmd_ctx):
    rendered_schemas = []
    search_schemas = {}
    discriminators = []
    if isinstance(schema, CMDObjectSchemaBase):

        if schema.discriminators:
            if schema.additional_props:
                raise NotImplementedError()
            discriminators.extend(schema.discriminators)

        if schema.props and schema.additional_props:
            # treat it as AAZFreeFormDictType
            pass
        elif schema.props:
            for s in schema.props:
                s_name = to_snake_case(s.name)
                s_typ, s_typ_kwargs, cls_builder_name = render_schema(s, cmd_ctx.response_clses, s_name)
                rendered_schemas.append((s_name, s_typ, s_typ_kwargs, cls_builder_name))
                if not cls_builder_name and isinstance(s, (CMDObjectSchemaBase, CMDArraySchemaBase)):
                    search_schemas[s_name] = s
        elif schema.additional_props:
            if schema.additional_props.item is not None:
                # AAZDictType
                s = schema.additional_props.item
                s_name = "Element"
                s_typ, s_typ_kwargs, cls_builder_name = render_schema_base(s, cmd_ctx.response_clses)
                rendered_schemas.append((s_name, s_typ, s_typ_kwargs, cls_builder_name))
                if not cls_builder_name and isinstance(s, (CMDObjectSchemaBase, CMDArraySchemaBase)):
                    search_schemas[s_name] = s
            else:
                assert schema.additional_props.any_type is True
        else:
            # allow empty object
            pass

    elif isinstance(schema, CMDObjectSchemaDiscriminator):
        if schema.discriminators:
            discriminators.extend(schema.discriminators)

        if schema.props:
            for s in schema.props:
                s_name = to_snake_case(s.name)
                s_typ, s_typ_kwargs, cls_builder_name = render_schema(s, cmd_ctx.response_clses, s_name)
                rendered_schemas.append((s_name, s_typ, s_typ_kwargs, cls_builder_name))
                if not cls_builder_name and isinstance(s, (CMDObjectSchemaBase, CMDArraySchemaBase)):
                    search_schemas[s_name] = s
        else:
            # allow empty object
            pass

    elif isinstance(schema, CMDArraySchemaBase):
        # AAZListType
        assert schema.item is not None
        s = schema.item
        s_name = "Element"
        s_typ, s_typ_kwargs, cls_builder_name = render_schema_base(s, cmd_ctx.response_clses)
        rendered_schemas.append((s_name, s_typ, s_typ_kwargs, cls_builder_name))
        if not cls_builder_name and isinstance(s, (CMDObjectSchemaBase, CMDArraySchemaBase)):
            search_schemas[s_name] = s

    else:
        raise NotImplementedError()

    if rendered_schemas:
        yield name, scope_define, rendered_schemas

    for s_name, s in search_schemas.items():
        s_scope_define = f"{scope_define}.{s_name}"
        if s_name == "Element":
            s_name = '_element'
        for scopes in _iter_response_scopes_by_schema_base(s, s_name, s_scope_define, cmd_ctx):
            yield scopes

    for disc in discriminators:
        key_name = to_snake_case(disc.property)
        key_value = disc.value
        disc_name = f"disc_{to_snake_case(disc.get_safe_value())}"

        disc_scope_define = f'{scope_define}.discriminate_by("{key_name}", "{key_value}")'
        for scopes in _iter_response_scopes_by_schema_base(disc, disc_name, disc_scope_define, cmd_ctx):
            yield scopes