(self, params, operation_model=None)
| 116 | return self._send_request(request_dict, operation_model) |
| 117 | |
| 118 | def create_request(self, params, operation_model=None): |
| 119 | request = create_request_object(params) |
| 120 | if operation_model: |
| 121 | request.stream_output = any( |
| 122 | [ |
| 123 | operation_model.has_streaming_output, |
| 124 | operation_model.has_event_stream_output, |
| 125 | ] |
| 126 | ) |
| 127 | service_id = operation_model.service_model.service_id.hyphenize() |
| 128 | event_name = f'request-created.{service_id}.{operation_model.name}' |
| 129 | self._event_emitter.emit( |
| 130 | event_name, |
| 131 | request=request, |
| 132 | operation_name=operation_model.name, |
| 133 | ) |
| 134 | prepared_request = self.prepare_request(request) |
| 135 | return prepared_request |
| 136 | |
| 137 | def _encode_headers(self, headers): |
| 138 | # In place encoding of headers to utf-8 if they are unicode. |
no test coverage detected