MCPcopy
hub / github.com/urllib3/urllib3 / get

Method get

src/urllib3/response.py:399–432  ·  src/urllib3/response.py::BytesQueueBuffer.get
(self, n: int)

Source from the content-addressed store, hash-verified

397 self._size += len(data)
398
399 def get(self, n: int) -> bytes:
400 if n == 0:
401 return bclass="st">""
402 elif not self.buffer:
403 raise RuntimeError(class="st">"buffer is empty")
404 elif n < 0:
405 raise ValueError(class="st">"n should be > 0")
406
407 if len(self.buffer[0]) == n and isinstance(self.buffer[0], bytes):
408 self._size -= n
409 return self.buffer.popleft()
410
411 fetched = 0
412 ret = io.BytesIO()
413 while fetched < n:
414 remaining = n - fetched
415 chunk = self.buffer.popleft()
416 chunk_length = len(chunk)
417 if remaining < chunk_length:
418 chunk = memoryview(chunk)
419 left_chunk, right_chunk = chunk[:remaining], chunk[remaining:]
420 ret.write(left_chunk)
421 self.buffer.appendleft(right_chunk)
422 self._size -= remaining
423 break
424 else:
425 ret.write(chunk)
426 self._size -= chunk_length
427 fetched += chunk_length
428
429 if not self.buffer:
430 break
431
432 return ret.getvalue()
433
434 def get_all(self) -> bytes:
435 buffer = self.buffer

Callers 15

test_single_chunkMethod · 0.95
test_read_too_muchMethod · 0.95
test_multiple_chunksMethod · 0.95
__init__.pyFile · 0.80
test_expireMethod · 0.80
test_getMethod · 0.80
test_pool_closeMethod · 0.80
test_contextmanagerMethod · 0.80
__call__Method · 0.80
test_fnFunction · 0.80

Calls

no outgoing calls

Tested by 15

test_single_chunkMethod · 0.76
test_read_too_muchMethod · 0.76
test_multiple_chunksMethod · 0.76
test_expireMethod · 0.64
test_getMethod · 0.64
test_pool_closeMethod · 0.64
test_contextmanagerMethod · 0.64
__call__Method · 0.64
test_fnFunction · 0.64
test_headersMethod · 0.64