(
self,
request: Optional[network.Request],
url: Optional[str],
method: str = None,
headers: Headers = None,
data: DataType = None,
params: ParamsType = None,
form: FormType = None,
multipart: MultipartType = None,
timeout: float = None,
failOnStatusCode: bool = None,
ignoreHTTPSErrors: bool = None,
maxRedirects: int = None,
maxRetries: int = None,
)
| 342 | ) |
| 343 | |
| 344 | async def _inner_fetch( |
| 345 | self, |
| 346 | request: Optional[network.Request], |
| 347 | url: Optional[str], |
| 348 | method: str = None, |
| 349 | headers: Headers = None, |
| 350 | data: DataType = None, |
| 351 | params: ParamsType = None, |
| 352 | form: FormType = None, |
| 353 | multipart: MultipartType = None, |
| 354 | timeout: float = None, |
| 355 | failOnStatusCode: bool = None, |
| 356 | ignoreHTTPSErrors: bool = None, |
| 357 | maxRedirects: int = None, |
| 358 | maxRetries: int = None, |
| 359 | ) -> "APIResponse": |
| 360 | if self._close_reason: |
| 361 | raise TargetClosedError(self._close_reason) |
| 362 | assert ( |
| 363 | (1 if data else 0) + (1 if form else 0) + (1 if multipart else 0) |
| 364 | ) <= 1, "Only one of 'data', 'form' or 'multipart' can be specified" |
| 365 | assert ( |
| 366 | maxRedirects is None or maxRedirects >= 0 |
| 367 | ), "'max_redirects' must be greater than or equal to '0'" |
| 368 | assert ( |
| 369 | maxRetries is None or maxRetries >= 0 |
| 370 | ), "'max_retries' must be greater than or equal to '0'" |
| 371 | url = url or (request.url if request else url) |
| 372 | method = method or (request.method if request else "GET") |
| 373 | # Cannot call allHeaders() here as the request may be paused inside route handler. |
| 374 | headers_obj = headers or (request.headers if request else None) |
| 375 | serialized_headers = serialize_headers(headers_obj) if headers_obj else None |
| 376 | json_data: Any = None |
| 377 | form_data: Optional[List[NameValue]] = None |
| 378 | multipart_data: Optional[List[FormField]] = None |
| 379 | post_data_buffer: Optional[bytes] = None |
| 380 | if data is not None: |
| 381 | if isinstance(data, str): |
| 382 | if is_json_content_type(serialized_headers): |
| 383 | json_data = data if is_json_parsable(data) else json.dumps(data) |
| 384 | else: |
| 385 | post_data_buffer = data.encode() |
| 386 | elif isinstance(data, bytes): |
| 387 | post_data_buffer = data |
| 388 | elif isinstance(data, (dict, list, int, bool)): |
| 389 | json_data = json.dumps(data) |
| 390 | else: |
| 391 | raise Error(f"Unsupported 'data' type: {type(data)}") |
| 392 | elif form: |
| 393 | if isinstance(form, FormData): |
| 394 | form_data = [] |
| 395 | for fd_name, fd_value in form._fields: |
| 396 | if isinstance(fd_value, (pathlib.Path, dict)): |
| 397 | raise Error( |
| 398 | f"Form field {fd_name!r} must be a string, number or boolean. Use 'multipart' for file uploads." |
| 399 | ) |
| 400 | form_data.append(NameValue(name=fd_name, value=str(fd_value))) |
| 401 | else: |
no test coverage detected