ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

PATH: opt / cloudlinux / venv / lib / python3.11 / site-packages / aiohttp

"""
Payload implementation for coroutines as data provider.

As a simple case, you can upload data from file::

@aiohttp.streamer
   async def file_sender(writer, file_name=None):
      with open(file_name, 'rb') as f:
          chunk = f.read(2**16)
          while chunk:
              await writer.write(chunk)

chunk = f.read(2**16)

Then you can use `file_sender` like this:

async with session.post('http://httpbin.org/post',
                            data=file_sender(file_name='huge_file')) as resp:
        print(await resp.text())

..note:: Coroutine must accept `writer` as first argument

"""

import types
import warnings
from typing import Any, Awaitable, Callable, Dict, Tuple

from .abc import AbstractStreamWriter
from .payload import Payload, payload_type

__all__ = ("streamer",)

class _stream_wrapper:
    def __init__(
        self,
        coro: Callable[..., Awaitable[None]],
        args: Tuple[Any, ...],
        kwargs: Dict[str, Any],
    ) -> None:
        self.coro = types.coroutine(coro)
        self.args = args
        self.kwargs = kwargs

async def __call__(self, writer: AbstractStreamWriter) -> None:
        await self.coro(writer, *self.args, **self.kwargs)

class streamer:
    def __init__(self, coro: Callable[..., Awaitable[None]]) -> None:
        warnings.warn(
            "@streamer is deprecated, use async generators instead",
            DeprecationWarning,
            stacklevel=2,
        )
        self.coro = coro

def __call__(self, *args: Any, **kwargs: Any) -> _stream_wrapper:
        return _stream_wrapper(self.coro, args, kwargs)

@payload_type(_stream_wrapper)
class StreamWrapperPayload(Payload):
    async def write(self, writer: AbstractStreamWriter) -> None:
        await self._value(writer)

@payload_type(streamer)
class StreamPayload(StreamWrapperPayload):
    def __init__(self, value: Any, *args: Any, **kwargs: Any) -> None:
        super().__init__(value(), *args, **kwargs)

async def write(self, writer: AbstractStreamWriter) -> None:
        await self._value(writer)

[-] _http_parser.pyx [open]
[-] http_writer.py [open]
[+] ..
[-] client_reqrep.py [open]
[-] http_exceptions.py [open]
[+] __pycache__
[-] web_routedef.py [open]
[-] streams.py [open]
[-] compression_utils.py [open]
[-] web_server.py [open]
[-] _http_writer.cpython-311-x86_64-linux-gnu.so [open]
[-] web.py [open]
[-] web_response.py [open]
[-] payload_streamer.py [open]
[-] pytest_plugin.py [open]
[-] _http_parser.cpython-311-x86_64-linux-gnu.so [open]
[-] connector.py [open]
[-] abc.py [open]
[-] resolver.py [open]
[-] _websocket.cpython-311-x86_64-linux-gnu.so [open]
[-] _cparser.pxd [open]
[-] log.py [open]
[-] client.py [open]
[-] cookiejar.py [open]
[-] web_ws.py [open]
[-] test_utils.py [open]
[-] _headers.pxi [open]
[-] worker.py [open]
[-] http_websocket.py [open]
[-] _find_header.pxd [open]
[-] client_exceptions.py [open]
[-] _helpers.cpython-311-x86_64-linux-gnu.so [open]
[-] base_protocol.py [open]
[-] tracing.py [open]
[-] formdata.py [open]
[-] _http_writer.pyx [open]
[-] _helpers.pyi [open]
[+] .hash
[-] web_fileresponse.py [open]
[-] helpers.py [open]
[-] http.py [open]
[-] web_exceptions.py [open]
[-] web_runner.py [open]
[-] web_app.py [open]
[-] payload.py [open]
[-] typedefs.py [open]
[-] tcp_helpers.py [open]
[-] _websocket.pyx [open]
[-] web_protocol.py [open]
[-] http_parser.py [open]
[-] __init__.py [open]
[-] client_proto.py [open]
[-] client_ws.py [open]
[-] multipart.py [open]
[-] web_log.py [open]
[-] web_middlewares.py [open]
[-] locks.py [open]
[-] hdrs.py [open]
[-] _helpers.pyx [open]
[-] web_urldispatcher.py [open]
[-] web_request.py [open]
[-] py.typed [open]