in src/aaz_dev/swagger/model/schema/operation.py [0:0]
def to_cmd(self, builder, parent_parameters, host_path, **kwargs):
cmd_op = CMDHttpOperation()
if self.x_ms_long_running_operation:
cmd_op.long_running = CMDHttpOperationLongRunning()
if self.x_ms_long_running_operation_options:
cmd_op.long_running.final_state_via = self.x_ms_long_running_operation_options.final_state_via
cmd_op.operation_id = self.operation_id
builder.setup_description(cmd_op, self)
cmd_op.http = CMDHttpAction()
if host_path:
cmd_op.http.path = host_path + builder.path
else:
cmd_op.http.path = builder.path
cmd_op.http.request = CMDHttpRequest()
cmd_op.http.request.method = builder.method
cmd_op.http.responses = []
# request
request = cmd_op.http.request
param_models = {}
client_request_id_name = None
if parent_parameters:
for p in parent_parameters:
model = builder(p)
if model is None:
continue
if p.IN_VALUE not in param_models:
param_models[p.IN_VALUE] = {}
param_models[p.IN_VALUE][p.name] = model
if p.IN_VALUE == HeaderParameter.IN_VALUE:
if p.x_ms_client_request_id or p.name == "x-ms-client-request-id":
client_request_id_name = p.name
if self.parameters:
for p in self.parameters:
model = builder(p)
if model is None:
continue
if p.IN_VALUE not in param_models:
param_models[p.IN_VALUE] = {}
param_models[p.IN_VALUE][p.name] = model
if p.IN_VALUE == HeaderParameter.IN_VALUE:
if p.x_ms_client_request_id or p.name == "x-ms-client-request-id":
client_request_id_name = p.name
if PathParameter.IN_VALUE in param_models:
request.path = CMDHttpRequestPath()
request.path.params = []
for _, model in sorted(param_models[PathParameter.IN_VALUE].items()):
request.path.params.append(model)
if QueryParameter.IN_VALUE in param_models:
request.query = CMDHttpRequestQuery()
request.query.params = []
for _, model in sorted(param_models[QueryParameter.IN_VALUE].items()):
request.query.params.append(model)
if HeaderParameter.IN_VALUE in param_models:
request.header = CMDHttpRequestHeader()
request.header.client_request_id = client_request_id_name
request.header.params = []
for name, model in sorted(param_models[HeaderParameter.IN_VALUE].items()):
if name == client_request_id_name:
# ignore client request id parameter for argument generation.
continue
request.header.params.append(model)
if BodyParameter.IN_VALUE in param_models:
if len(param_models[BodyParameter.IN_VALUE]) > 1:
raise exceptions.InvalidSwaggerValueError(
msg="Duplicate parameters in request body",
key=self.traces,
value=[builder.path, builder.method, builder.mutability, *param_models[BodyParameter.IN_VALUE].keys()]
)
model = [*param_models[BodyParameter.IN_VALUE].values()][0]
if isinstance(model, CMDRequestJson):
request.body = CMDHttpRequestJsonBody()
request.body.json = model
else:
raise NotImplementedError()
if FormDataParameter.IN_VALUE in param_models:
raise NotImplementedError()
# response
success_responses, redirect_responses, error_responses = builder.classify_responses(self)
cmd_op.http.responses = []
for status_codes, resp in success_responses:
model = builder(resp, status_codes=status_codes)
cmd_op.http.responses.append(model)
for status_codes, resp in redirect_responses:
model = builder(resp, status_codes=status_codes)
cmd_op.http.responses.append(model)
for status_codes, resp in error_responses:
model = builder(resp, is_error=True, status_codes=status_codes)
cmd_op.http.responses.append(model)
if len(cmd_op.http.responses) == 0:
raise exceptions.InvalidSwaggerValueError(
msg="No http response",
key=self.traces,
value=[builder.path, builder.method]
)
return cmd_op