in src/aaz_dev/swagger/model/schema/cmd_builder.py [0:0]
def classify_responses(schema):
success_responses = []
success_202_response = None
success_204_response = None
redirect_responses = []
error_responses = []
if 'default' in schema.responses:
resp = schema.responses['default']
error_responses.append((None, resp))
for code, resp in schema.responses.items():
if code == "default":
continue
status_code = int(code)
if status_code < 300:
if status_code == 202:
success_202_response = ({status_code}, resp)
elif status_code == 204:
success_204_response = ({status_code}, resp)
else:
# success
find_match = False
for status_codes, p_resp in success_responses:
if p_resp.schema == resp.schema:
status_codes.add(status_code)
find_match = True
break
if not find_match:
success_responses.append(({status_code}, resp))
elif status_code < 400:
# redirect
find_match = False
for status_codes, p_resp in redirect_responses:
if p_resp.schema == resp.schema:
status_codes.add(status_code)
find_match = True
break
if not find_match:
redirect_responses.append(({status_code}, resp))
else:
# error
find_match = False
for status_codes, p_resp in error_responses:
if p_resp.schema == resp.schema:
if status_codes is not None: # ignore to add into default response
status_codes.add(status_code)
find_match = True
if not find_match:
error_responses.append(({status_code}, resp))
if len(success_responses) >= 2:
raise exceptions.InvalidSwaggerValueError(
msg="Multi Schema for success responses",
key=[schema.traces],
value=[status_codes for status_codes, _ in success_responses]
)
if len(redirect_responses) >= 2:
raise exceptions.InvalidSwaggerValueError(
msg="Multi Schema for redirect responses",
key=[schema.traces],
value=[status_codes for status_codes, _ in redirect_responses]
)
if len(error_responses) >= 3:
raise exceptions.InvalidSwaggerValueError(
msg="Multi Schema for error responses",
key=[schema.traces],
value=[status_codes for status_codes, _ in error_responses]
)
if success_202_response is not None:
# append 202 Long Running response at the end of success response
success_responses.append(success_202_response)
if success_204_response is not None:
# append 204 No Content response at the end of success response
success_responses.append(success_204_response)
success_codes = reduce(lambda x, y: x | y, [codes for codes, _ in success_responses], set())
if schema.x_ms_long_running_operation and not success_codes & {200, 201}:
if lro_schema := schema.x_ms_lro_final_state_schema:
lro_response = Response()
lro_response.description = "Response schema for long-running operation."
lro_response.schema = lro_schema
success_responses.append(({200, 201}, lro_response)) # use `final-state-schema` as response
elif operation_id_separate(schema.operation_id)[-1][0] == "delete":
lro_response = Response()
lro_response.description = "Response schema for long-running operation."
success_responses.append(({200, 201}, lro_response))
else:
logger.warning(f"No response schema for long-running-operation: {schema.operation_id}.")
# # default response
# if 'default' not in error_responses and len(error_responses) == 1:
# p_resp, p_model = [*error_responses.values()][0]
# if p_model.body is not None:
# # use the current error response as default
# p_model.status_codes = None
# error_responses = {
# "default": (p_resp, p_model)
# }
# if 'default' not in error_responses:
# raise exceptions.InvalidSwaggerValueError(
# msg="Miss default response",
# key=self.traces,
# value=[path, method]
# )
return success_responses, redirect_responses, error_responses