def __init__()

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

• 69 lines of code
• 21 McCabe index (conditional complexity)

    def __init__(self, name, cmd_ctx, operation, client_endpoints):
        super().__init__(name, cmd_ctx, operation)
        assert isinstance(self._operation, CMDHttpOperation)

        self._client_endpoints = client_endpoints

        if self._operation.long_running is not None:
            self.is_long_running = True
            self.lro_options = {
                'final-state-via': self._operation.long_running.final_state_via
            }

        self.success_responses = []
        self.success_202_response = None

        error_format = None
        for response in self._operation.http.responses:
            if not response.is_error:
                if self.is_long_running and response.status_codes == [202]:
                    # ignore 202 response for long running operation.
                    # Long running operation should use 200 or 201 schema to deserialize the final response.
                    continue
                self.success_responses.append(AzHttpResponseGenerator(self._cmd_ctx, response))
            else:
                if not isinstance(response.body, CMDHttpResponseJsonBody):
                    if not response.body:
                        raise exceptions.InvalidAPIUsage(
                            f"Invalid `Error` response schema in operation `{self._operation.operation_id}`: "
                            f"Missing `schema` property in response "
                            f"`{response.status_codes or 'default'}`."
                        )
                    else:
                        raise exceptions.InvalidAPIUsage(
                            f"Invalid `Error` response schema in operation `{self._operation.operation_id}`: "
                            f"Only support json schema, current is '{type(response.body)}' in response "
                            f"`{response.status_codes or 'default'}`"
                        )
                schema = response.body.json.schema
                if not isinstance(schema, CMDClsSchemaBase):
                    raise NotImplementedError()
                name = schema.type[1:]
                if not error_format:
                    error_format = name
                if error_format != name:
                    raise exceptions.InvalidAPIUsage(
                        f"Invalid `Error` response schema in operation `{self._operation.operation_id}`: "
                        f"Multiple schema formats are founded: {name}, {error_format}"
                    )
        if not error_format:
            raise exceptions.InvalidAPIUsage(
                f"Missing `Error` response schema in operation `{self._operation.operation_id}`: "
                f"Please define the `default` response in swagger for error."
            )
        elif not AAZErrorFormatEnum.validate(error_format):
            raise exceptions.InvalidAPIUsage(
                f"Invalid `Error` response schema in operation `{self._operation.operation_id}`: "
                f"Invalid error format `{error_format}`. Support `ODataV4Format` and `MgmtErrorFormat` only"
            )

        if self.is_long_running:
            callback_name = None
            for response in self.success_responses:
                if 200 in response.status_codes or 201 in response.status_codes:
                    callback_name = response.callback_name
                    break
            self.success_202_response = AzHttp202ResponseGenerator(self._cmd_ctx, callback_name)
        self.error_format = error_format

        # specify content
        self.content = None
        self.form_content = None
        self.stream_content = None
        if self._operation.http.request.body:
            body = self._operation.http.request.body
            if isinstance(body, CMDHttpRequestJsonBody):
                self.content = AzHttpRequestContentGenerator(self._cmd_ctx, body)
            else:
                raise NotImplementedError()