import asyncio import enum import io import json import mimetypes import os import sys import warnings from abc import ABC, abstractmethod from collections.abc import Iterable from itertools import chain from typing import ( IO, TYPE_CHECKING, Any, Dict, Final, List, Optional, Set, TextIO, Tuple, Type, Union, ) from multidict import CIMultiDict from . import hdrs from .abc import AbstractStreamWriter from .helpers import ( _SENTINEL, content_disposition_header, guess_filename, parse_mimetype, sentinel, ) from .streams import StreamReader from .typedefs import JSONEncoder, _CIMultiDict __all__ = ( "PAYLOAD_REGISTRY", "get_payload", "payload_type", "Payload", "BytesPayload", "StringPayload", "IOBasePayload", "BytesIOPayload", "BufferedReaderPayload", "TextIOPayload", "StringIOPayload", "JsonPayload", "AsyncIterablePayload", ) TOO_LARGE_BYTES_BODY: Final[int] = 2**20 # 1 MB READ_SIZE: Final[int] = 2**16 # 64 KB _CLOSE_FUTURES: Set[asyncio.Future[None]] = set() class LookupError(Exception): """Raised when no payload factory is found for the given data type.""" class Order(str, enum.Enum): normal = "normal" try_first = "try_first" try_last = "try_last" def get_payload(data: Any, *args: Any, **kwargs: Any) -> "Payload": return PAYLOAD_REGISTRY.get(data, *args, **kwargs) def register_payload( factory: Type["Payload"], type: Any, *, order: Order = Order.normal ) -> None: PAYLOAD_REGISTRY.register(factory, type, order=order) class payload_type: def __init__(self, type: Any, *, order: Order = Order.normal) -> None: self.type = type self.order = order def __call__(self, factory: Type["Payload"]) -> Type["Payload"]: register_payload(factory, self.type, order=self.order) return factory PayloadType = Type["Payload"] _PayloadRegistryItem = Tuple[PayloadType, Any] class PayloadRegistry: """Payload registry. note: we need zope.interface for more efficient adapter search """ __slots__ = ("_first", "_normal", "_last", "_normal_lookup") def __init__(self) -> None: self._first: List[_PayloadRegistryItem] = [] self._normal: List[_PayloadRegistryItem] = [] self._last: List[_PayloadRegistryItem] = [] self._normal_lookup: Dict[Any, PayloadType] = {} def get( self, data: Any, *args: Any, _CHAIN: "Type[chain[_PayloadRegistryItem]]" = chain, **kwargs: Any, ) -> "Payload": if self._first: for factory, type_ in self._first: if isinstance(data, type_): return factory(data, *args, **kwargs) # Try the fast lookup first if lookup_factory := self._normal_lookup.get(type(data)): return lookup_factory(data, *args, **kwargs) # Bail early if its already a Payload if isinstance(data, Payload): return data # Fallback to the slower linear search for factory, type_ in _CHAIN(self._normal, self._last): if isinstance(data, type_): return factory(data, *args, **kwargs) raise LookupError() def register( self, factory: PayloadType, type: Any, *, order: Order = Order.normal ) -> None: if order is Order.try_first: self._first.append((factory, type)) elif order is Order.normal: self._normal.append((factory, type)) if isinstance(type, Iterable): for t in type: self._normal_lookup[t] = factory else: self._normal_lookup[type] = factory elif order is Order.try_last: self._last.append((factory, type)) else: raise ValueError(f"Unsupported order {order!r}") class Payload(ABC): _default_content_type: str = "application/octet-stream" _size: Optional[int] = None _consumed: bool = False # Default: payload has not been consumed yet _autoclose: bool = False # Default: assume resource needs explicit closing def __init__( self, value: Any, headers: Optional[ Union[_CIMultiDict, Dict[str, str], Iterable[Tuple[str, str]]] ] = None, content_type: Union[str, None, _SENTINEL] = sentinel, filename: Optional[str] = None, encoding: Optional[str] = None, **kwargs: Any, ) -> None: self._encoding = encoding self._filename = filename self._headers: _CIMultiDict = CIMultiDict() self._value = value if content_type is not sentinel and content_type is not None: self._headers[hdrs.CONTENT_TYPE] = content_type elif self._filename is not None: if sys.version_info >= (3, 13): guesser = mimetypes.guess_file_type else: guesser = mimetypes.guess_type content_type = guesser(self._filename)[0] if content_type is None: content_type = self._default_content_type self._headers[hdrs.CONTENT_TYPE] = content_type else: self._headers[hdrs.CONTENT_TYPE] = self._default_content_type if headers: self._headers.update(headers) @property def size(self) -> Optional[int]: """Size of the payload in bytes. Returns the number of bytes that will be transmitted when the payload is written. For string payloads, this is the size after encoding to bytes, not the length of the string. """ return self._size @property def filename(self) -> Optional[str]: """Filename of the payload.""" return self._filename @property def headers(self) -> _CIMultiDict: """Custom item headers""" return self._headers @property def _binary_headers(self) -> bytes: return ( "".join([k + ": " + v + "\r\n" for k, v in self.headers.items()]).encode( "utf-8" ) + b"\r\n" ) @property def encoding(self) -> Optional[str]: """Payload encoding""" return self._encoding @property def content_type(self) -> str: """Content type""" return self._headers[hdrs.CONTENT_TYPE] @property def consumed(self) -> bool: """Whether the payload has been consumed and cannot be reused.""" return self._consumed @property def autoclose(self) -> bool: """ Whether the payload can close itself automatically. Returns True if the payload has no file handles or resources that need explicit closing. If False, callers must await close() to release resources. """ return self._autoclose def set_content_disposition( self, disptype: str, quote_fields: bool = True, _charset: str = "utf-8", **params: Any, ) -> None: """Sets ``Content-Disposition`` header.""" self._headers[hdrs.CONTENT_DISPOSITION] = content_disposition_header( disptype, quote_fields=quote_fields, _charset=_charset, **params ) @abstractmethod def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str: """ Return string representation of the value. This is named decode() to allow compatibility with bytes objects. """ @abstractmethod async def write(self, writer: AbstractStreamWriter) -> None: """ Write payload to the writer stream. Args: writer: An AbstractStreamWriter instance that handles the actual writing This is a legacy method that writes the entire payload without length constraints. Important: For new implementations, use write_with_length() instead of this method. This method is maintained for backwards compatibility and will eventually delegate to write_with_length(writer, None) in all implementations. All payload subclasses must override this method for backwards compatibility, but new code should use write_with_length for more flexibility and control. """ # write_with_length is new in aiohttp 3.12 # it should be overridden by subclasses async def write_with_length( self, writer: AbstractStreamWriter, content_length: Optional[int] ) -> None: """ Write payload with a specific content length constraint. Args: writer: An AbstractStreamWriter instance that handles the actual writing content_length: Maximum number of bytes to write (None for unlimited) This method allows writing payload content with a specific length constraint, which is particularly useful for HTTP responses with Content-Length header. Note: This is the base implementation that provides backwards compatibility for subclasses that don't override this method. Specific payload types should override this method to implement proper length-constrained writing. """ # Backwards compatibility for subclasses that don't override this method # and for the default implementation await self.write(writer) async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes: """ Return bytes representation of the value. This is a convenience method that calls decode() and encodes the result to bytes using the specified encoding. """ # Use instance encoding if available, otherwise use parameter actual_encoding = self._encoding or encoding return self.decode(actual_encoding, errors).encode(actual_encoding) def _close(self) -> None: """ Async safe synchronous close operations for backwards compatibility. This method exists only for backwards compatibility with code that needs to clean up payloads synchronously. In the future, we will drop this method and only support the async close() method. WARNING: This method must be safe to call from within the event loop without blocking. Subclasses should not perform any blocking I/O here. WARNING: This method must be called from within an event loop for certain payload types (e.g., IOBasePayload). Calling it outside an event loop may raise RuntimeError. """ # This is a no-op by default, but subclasses can override it # for non-blocking cleanup operations. async def close(self) -> None: """ Close the payload if it holds any resources. IMPORTANT: This method must not await anything that might not finish immediately, as it may be called during cleanup/cancellation. Schedule any long-running operations without awaiting them. In the future, this will be the only close method supported. """ self._close() class BytesPayload(Payload): _value: bytes # _consumed = False (inherited) - Bytes are immutable and can be reused _autoclose = True # No file handle, just bytes in memory def __init__( self, value: Union[bytes, bytearray, memoryview], *args: Any, **kwargs: Any ) -> None: if "content_type" not in kwargs: kwargs["content_type"] = "application/octet-stream" super().__init__(value, *args, **kwargs) if isinstance(value, memoryview): self._size = value.nbytes elif isinstance(value, (bytes, bytearray)): self._size = len(value) else: raise TypeError(f"value argument must be byte-ish, not {type(value)!r}") if self._size > TOO_LARGE_BYTES_BODY: kwargs = {"source": self} warnings.warn( "Sending a large body directly with raw bytes might" " lock the event loop. You should probably pass an " "io.BytesIO object instead", ResourceWarning, **kwargs, ) def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str: return self._value.decode(encoding, errors) async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes: """ Return bytes representation of the value. This method returns the raw bytes content of the payload. It is equivalent to accessing the _value attribute directly. """ return self._value async def write(self, writer: AbstractStreamWriter) -> None: """ Write the entire bytes payload to the writer stream. Args: writer: An AbstractStreamWriter instance that handles the actual writing This method writes the entire bytes content without any length constraint. Note: For new implementations that need length control, use write_with_length(). This method is maintained for backwards compatibility and is equivalent to write_with_length(writer, None). """ await writer.write(self._value) async def write_with_length( self, writer: AbstractStreamWriter, content_length: Optional[int] ) -> None: """ Write bytes payload with a specific content length constraint. Args: writer: An AbstractStreamWriter instance that handles the actual writing content_length: Maximum number of bytes to write (None for unlimited) This method writes either the entire byte sequence or a slice of it up to the specified content_length. For BytesPayload, this operation is performed efficiently using array slicing. """ if content_length is not None: await writer.write(self._value[:content_length]) else: await writer.write(self._value) class StringPayload(BytesPayload): def __init__( self, value: str, *args: Any, encoding: Optional[str] = None, content_type: Optional[str] = None, **kwargs: Any, ) -> None: if encoding is None: if content_type is None: real_encoding = "utf-8" content_type = "text/plain; charset=utf-8" else: mimetype = parse_mimetype(content_type) real_encoding = mimetype.parameters.get("charset", "utf-8") else: if content_type is None: content_type = "text/plain; charset=%s" % encoding real_encoding = encoding super().__init__( value.encode(real_encoding), encoding=real_encoding, content_type=content_type, *args, **kwargs, ) class StringIOPayload(StringPayload): def __init__(self, value: IO[str], *args: Any, **kwargs: Any) -> None: super().__init__(value.read(), *args, **kwargs) class IOBasePayload(Payload): _value: io.IOBase # _consumed = False (inherited) - File can be re-read from the same position _start_position: Optional[int] = None # _autoclose = False (inherited) - Has file handle that needs explicit closing def __init__( self, value: IO[Any], disposition: str = "attachment", *args: Any, **kwargs: Any ) -> None: if "filename" not in kwargs: kwargs["filename"] = guess_filename(value) super().__init__(value, *args, **kwargs) if self._filename is not None and disposition is not None: if hdrs.CONTENT_DISPOSITION not in self.headers: self.set_content_disposition(disposition, filename=self._filename) def _set_or_restore_start_position(self) -> None: """Set or restore the start position of the file-like object.""" if self._start_position is None: try: self._start_position = self._value.tell() except OSError: self._consumed = True # Cannot seek, mark as consumed return self._value.seek(self._start_position) def _read_and_available_len( self, remaining_content_len: Optional[int] ) -> Tuple[Optional[int], bytes]: """ Read the file-like object and return both its total size and the first chunk. Args: remaining_content_len: Optional limit on how many bytes to read in this operation. If None, READ_SIZE will be used as the default chunk size. Returns: A tuple containing: - The total size of the remaining unread content (None if size cannot be determined) - The first chunk of bytes read from the file object This method is optimized to perform both size calculation and initial read in a single operation, which is executed in a single executor job to minimize context switches and file operations when streaming content. """ self._set_or_restore_start_position() size = self.size # Call size only once since it does I/O return size, self._value.read( min(READ_SIZE, size or READ_SIZE, remaining_content_len or READ_SIZE) ) def _read(self, remaining_content_len: Optional[int]) -> bytes: """ Read a chunk of data from the file-like object. Args: remaining_content_len: Optional maximum number of bytes to read. If None, READ_SIZE will be used as the default chunk size. Returns: A chunk of bytes read from the file object, respecting the remaining_content_len limit if specified. This method is used for subsequent reads during streaming after the initial _read_and_available_len call has been made. """ return self._value.read(remaining_content_len or READ_SIZE) # type: ignore[no-any-return] @property def size(self) -> Optional[int]: """ Size of the payload in bytes. Returns the number of bytes remaining to be read from the file. Returns None if the size cannot be determined (e.g., for unseekable streams). """ try: return os.fstat(self._value.fileno()).st_size - self._value.tell() except (AttributeError, OSError): return None async def write(self, writer: AbstractStreamWriter) -> None: """ Write the entire file-like payload to the writer stream. Args: writer: An AbstractStreamWriter instance that handles the actual writing This method writes the entire file content without any length constraint. It delegates to write_with_length() with no length limit for implementation consistency. Note: For new implementations that need length control, use write_with_length() directly. This method is maintained for backwards compatibility with existing code. """ await self.write_with_length(writer, None) async def write_with_length( self, writer: AbstractStreamWriter, content_length: Optional[int] ) -> None: """ Write file-like payload with a specific content length constraint. Args: writer: An AbstractStreamWriter instance that handles the actual writing content_length: Maximum number of bytes to write (None for unlimited) This method implements optimized streaming of file content with length constraints: 1. File reading is performed in a thread pool to avoid blocking the event loop 2. Content is read and written in chunks to maintain memory efficiency 3. Writing stops when either: - All available file content has been written (when size is known) - The specified content_length has been reached 4. File resources are properly closed even if the operation is cancelled The implementation carefully handles both known-size and unknown-size payloads, as well as constrained and unconstrained content lengths. """ loop = asyncio.get_running_loop() total_written_len = 0 remaining_content_len = content_length # Get initial data and available length available_len, chunk = await loop.run_in_executor( None, self._read_and_available_len, remaining_content_len ) # Process data chunks until done while chunk: chunk_len = len(chunk) # Write data with or without length constraint if remaining_content_len is None: await writer.write(chunk) else: await writer.write(chunk[:remaining_content_len]) remaining_content_len -= chunk_len total_written_len += chunk_len # Check if we're done writing if self._should_stop_writing( available_len, total_written_len, remaining_content_len ): return # Read next chunk chunk = await loop.run_in_executor( None, self._read, ( min(READ_SIZE, remaining_content_len) if remaining_content_len is not None else READ_SIZE ), ) def _should_stop_writing( self, available_len: Optional[int], total_written_len: int, remaining_content_len: Optional[int], ) -> bool: """ Determine if we should stop writing data. Args: available_len: Known size of the payload if available (None if unknown) total_written_len: Number of bytes already written remaining_content_len: Remaining bytes to be written for content-length limited responses Returns: True if we should stop writing data, based on either: - Having written all available data (when size is known) - Having written all requested content (when content-length is specified) """ return (available_len is not None and total_written_len >= available_len) or ( remaining_content_len is not None and remaining_content_len <= 0 ) def _close(self) -> None: """ Async safe synchronous close operations for backwards compatibility. This method exists only for backwards compatibility. Use the async close() method instead. WARNING: This method MUST be called from within an event loop. Calling it outside an event loop will raise RuntimeError. """ # Skip if already consumed if self._consumed: return self._consumed = True # Mark as consumed to prevent further writes # Schedule file closing without awaiting to prevent cancellation issues loop = asyncio.get_running_loop() close_future = loop.run_in_executor(None, self._value.close) # Hold a strong reference to the future to prevent it from being # garbage collected before it completes. _CLOSE_FUTURES.add(close_future) close_future.add_done_callback(_CLOSE_FUTURES.remove) async def close(self) -> None: """ Close the payload if it holds any resources. IMPORTANT: This method must not await anything that might not finish immediately, as it may be called during cleanup/cancellation. Schedule any long-running operations without awaiting them. """ self._close() def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str: """ Return string representation of the value. WARNING: This method does blocking I/O and should not be called in the event loop. """ return self._read_all().decode(encoding, errors) def _read_all(self) -> bytes: """Read the entire file-like object and return its content as bytes.""" self._set_or_restore_start_position() # Use readlines() to ensure we get all content return b"".join(self._value.readlines()) async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes: """ Return bytes representation of the value. This method reads the entire file content and returns it as bytes. It is equivalent to reading the file-like object directly. The file reading is performed in an executor to avoid blocking the event loop. """ loop = asyncio.get_running_loop() return await loop.run_in_executor(None, self._read_all) class TextIOPayload(IOBasePayload): _value: io.TextIOBase # _autoclose = False (inherited) - Has text file handle that needs explicit closing def __init__( self, value: TextIO, *args: Any, encoding: Optional[str] = None, content_type: Optional[str] = None, **kwargs: Any, ) -> None: if encoding is None: if content_type is None: encoding = "utf-8" content_type = "text/plain; charset=utf-8" else: mimetype = parse_mimetype(content_type) encoding = mimetype.parameters.get("charset", "utf-8") else: if content_type is None: content_type = "text/plain; charset=%s" % encoding super().__init__( value, content_type=content_type, encoding=encoding, *args, **kwargs, ) def _read_and_available_len( self, remaining_content_len: Optional[int] ) -> Tuple[Optional[int], bytes]: """ Read the text file-like object and return both its total size and the first chunk. Args: remaining_content_len: Optional limit on how many bytes to read in this operation. If None, READ_SIZE will be used as the default chunk size. Returns: A tuple containing: - The total size of the remaining unread content (None if size cannot be determined) - The first chunk of bytes read from the file object, encoded using the payload's encoding This method is optimized to perform both size calculation and initial read in a single operation, which is executed in a single executor job to minimize context switches and file operations when streaming content. Note: TextIOPayload handles encoding of the text content before writing it to the stream. If no encoding is specified, UTF-8 is used as the default. """ self._set_or_restore_start_position() size = self.size chunk = self._value.read( min(READ_SIZE, size or READ_SIZE, remaining_content_len or READ_SIZE) ) return size, chunk.encode(self._encoding) if self._encoding else chunk.encode() def _read(self, remaining_content_len: Optional[int]) -> bytes: """ Read a chunk of data from the text file-like object. Args: remaining_content_len: Optional maximum number of bytes to read. If None, READ_SIZE will be used as the default chunk size. Returns: A chunk of bytes read from the file object and encoded using the payload's encoding. The data is automatically converted from text to bytes. This method is used for subsequent reads during streaming after the initial _read_and_available_len call has been made. It properly handles text encoding, converting the text content to bytes using the specified encoding (or UTF-8 if none was provided). """ chunk = self._value.read(remaining_content_len or READ_SIZE) return chunk.encode(self._encoding) if self._encoding else chunk.encode() def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str: """ Return string representation of the value. WARNING: This method does blocking I/O and should not be called in the event loop. """ self._set_or_restore_start_position() return self._value.read() async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes: """ Return bytes representation of the value. This method reads the entire text file content and returns it as bytes. It encodes the text content using the specified encoding. The file reading is performed in an executor to avoid blocking the event loop. """ loop = asyncio.get_running_loop() # Use instance encoding if available, otherwise use parameter actual_encoding = self._encoding or encoding def _read_and_encode() -> bytes: self._set_or_restore_start_position() # TextIO read() always returns the full content return self._value.read().encode(actual_encoding, errors) return await loop.run_in_executor(None, _read_and_encode) class BytesIOPayload(IOBasePayload): _value: io.BytesIO _size: int # Always initialized in __init__ _autoclose = True # BytesIO is in-memory, safe to auto-close def __init__(self, value: io.BytesIO, *args: Any, **kwargs: Any) -> None: super().__init__(value, *args, **kwargs) # Calculate size once during initialization self._size = len(self._value.getbuffer()) - self._value.tell() @property def size(self) -> int: """Size of the payload in bytes. Returns the number of bytes in the BytesIO buffer that will be transmitted. This is calculated once during initialization for efficiency. """ return self._size def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str: self._set_or_restore_start_position() return self._value.read().decode(encoding, errors) async def write(self, writer: AbstractStreamWriter) -> None: return await self.write_with_length(writer, None) async def write_with_length( self, writer: AbstractStreamWriter, content_length: Optional[int] ) -> None: """ Write BytesIO payload with a specific content length constraint. Args: writer: An AbstractStreamWriter instance that handles the actual writing content_length: Maximum number of bytes to write (None for unlimited) This implementation is specifically optimized for BytesIO objects: 1. Reads content in chunks to maintain memory efficiency 2. Yields control back to the event loop periodically to prevent blocking when dealing with large BytesIO objects 3. Respects content_length constraints when specified 4. Properly cleans up by closing the BytesIO object when done or on error The periodic yielding to the event loop is important for maintaining responsiveness when processing large in-memory buffers. """ self._set_or_restore_start_position() loop_count = 0 remaining_bytes = content_length while chunk := self._value.read(READ_SIZE): if loop_count > 0: # Avoid blocking the event loop # if they pass a large BytesIO object # and we are not in the first iteration # of the loop await asyncio.sleep(0) if remaining_bytes is None: await writer.write(chunk) else: await writer.write(chunk[:remaining_bytes]) remaining_bytes -= len(chunk) if remaining_bytes <= 0: return loop_count += 1 async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes: """ Return bytes representation of the value. This method reads the entire BytesIO content and returns it as bytes. It is equivalent to accessing the _value attribute directly. """ self._set_or_restore_start_position() return self._value.read() async def close(self) -> None: """ Close the BytesIO payload. This does nothing since BytesIO is in-memory and does not require explicit closing. """ class BufferedReaderPayload(IOBasePayload): _value: io.BufferedIOBase # _autoclose = False (inherited) - Has buffered file handle that needs explicit closing def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str: self._set_or_restore_start_position() return self._value.read().decode(encoding, errors) class JsonPayload(BytesPayload): def __init__( self, value: Any, encoding: str = "utf-8", content_type: str = "application/json", dumps: JSONEncoder = json.dumps, *args: Any, **kwargs: Any, ) -> None: super().__init__( dumps(value).encode(encoding), content_type=content_type, encoding=encoding, *args, **kwargs, ) if TYPE_CHECKING: from typing import AsyncIterable, AsyncIterator _AsyncIterator = AsyncIterator[bytes] _AsyncIterable = AsyncIterable[bytes] else: from collections.abc import AsyncIterable, AsyncIterator _AsyncIterator = AsyncIterator _AsyncIterable = AsyncIterable class AsyncIterablePayload(Payload): _iter: Optional[_AsyncIterator] = None _value: _AsyncIterable _cached_chunks: Optional[List[bytes]] = None # _consumed stays False to allow reuse with cached content _autoclose = True # Iterator doesn't need explicit closing def __init__(self, value: _AsyncIterable, *args: Any, **kwargs: Any) -> None: if not isinstance(value, AsyncIterable): raise TypeError( "value argument must support " "collections.abc.AsyncIterable interface, " "got {!r}".format(type(value)) ) if "content_type" not in kwargs: kwargs["content_type"] = "application/octet-stream" super().__init__(value, *args, **kwargs) self._iter = value.__aiter__() async def write(self, writer: AbstractStreamWriter) -> None: """ Write the entire async iterable payload to the writer stream. Args: writer: An AbstractStreamWriter instance that handles the actual writing This method iterates through the async iterable and writes each chunk to the writer without any length constraint. Note: For new implementations that need length control, use write_with_length() directly. This method is maintained for backwards compatibility with existing code. """ await self.write_with_length(writer, None) async def write_with_length( self, writer: AbstractStreamWriter, content_length: Optional[int] ) -> None: """ Write async iterable payload with a specific content length constraint. Args: writer: An AbstractStreamWriter instance that handles the actual writing content_length: Maximum number of bytes to write (None for unlimited) This implementation handles streaming of async iterable content with length constraints: 1. If cached chunks are available, writes from them 2. Otherwise iterates through the async iterable one chunk at a time 3. Respects content_length constraints when specified 4. Does NOT generate cache - that's done by as_bytes() """ # If we have cached chunks, use them if self._cached_chunks is not None: remaining_bytes = content_length for chunk in self._cached_chunks: if remaining_bytes is None: await writer.write(chunk) elif remaining_bytes > 0: await writer.write(chunk[:remaining_bytes]) remaining_bytes -= len(chunk) else: break return # If iterator is exhausted and we don't have cached chunks, nothing to write if self._iter is None: return # Stream from the iterator remaining_bytes = content_length try: while True: if sys.version_info >= (3, 10): chunk = await anext(self._iter) else: chunk = await self._iter.__anext__() if remaining_bytes is None: await writer.write(chunk) # If we have a content length limit elif remaining_bytes > 0: await writer.write(chunk[:remaining_bytes]) remaining_bytes -= len(chunk) # We still want to exhaust the iterator even # if we have reached the content length limit # since the file handle may not get closed by # the iterator if we don't do this except StopAsyncIteration: # Iterator is exhausted self._iter = None self._consumed = True # Mark as consumed when streamed without caching def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str: """Decode the payload content as a string if cached chunks are available.""" if self._cached_chunks is not None: return b"".join(self._cached_chunks).decode(encoding, errors) raise TypeError("Unable to decode - content not cached. Call as_bytes() first.") async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes: """ Return bytes representation of the value. This method reads the entire async iterable content and returns it as bytes. It generates and caches the chunks for future reuse. """ # If we have cached chunks, return them joined if self._cached_chunks is not None: return b"".join(self._cached_chunks) # If iterator is exhausted and no cache, return empty if self._iter is None: return b"" # Read all chunks and cache them chunks: List[bytes] = [] async for chunk in self._iter: chunks.append(chunk) # Iterator is exhausted, cache the chunks self._iter = None self._cached_chunks = chunks # Keep _consumed as False to allow reuse with cached chunks return b"".join(chunks) class StreamReaderPayload(AsyncIterablePayload): def __init__(self, value: StreamReader, *args: Any, **kwargs: Any) -> None: super().__init__(value.iter_any(), *args, **kwargs) PAYLOAD_REGISTRY = PayloadRegistry() PAYLOAD_REGISTRY.register(BytesPayload, (bytes, bytearray, memoryview)) PAYLOAD_REGISTRY.register(StringPayload, str) PAYLOAD_REGISTRY.register(StringIOPayload, io.StringIO) PAYLOAD_REGISTRY.register(TextIOPayload, io.TextIOBase) PAYLOAD_REGISTRY.register(BytesIOPayload, io.BytesIO) PAYLOAD_REGISTRY.register(BufferedReaderPayload, (io.BufferedReader, io.BufferedRandom)) PAYLOAD_REGISTRY.register(IOBasePayload, io.IOBase) PAYLOAD_REGISTRY.register(StreamReaderPayload, StreamReader) # try_last for giving a chance to more specialized async interables like # multipart.BodyPartReaderPayload override the default PAYLOAD_REGISTRY.register(AsyncIterablePayload, AsyncIterable, order=Order.try_last)