class DeadlineWrapper(Wrapper):
    """Deadline wrapper to specify deadline once for any number of awaiting
    method calls.

    Example:

    .. code-block:: python3

        dw = DeadlineWrapper()

        with dw.start(deadline):
            await handle_request()

        # somewhere during request handling:

        async def blocking_call():
            with dw:
                await asyncio.sleep(10)

    """
    @contextmanager
    def start(self, deadline: 'Deadline') -> Iterator[None]:
        timeout = deadline.time_remaining()
        if not timeout:
            raise asyncio.TimeoutError('Deadline exceeded')

        def callback() -> None:
            self.cancel(asyncio.TimeoutError('Deadline exceeded'))

        loop = asyncio.get_event_loop()
        timer = loop.call_later(timeout, callback)
        try:
            yield
        finally:
            timer.cancel() class Wrapper(ContextManager[None]):
    """Special wrapper for coroutines to wake them up in case of some error.

    Example:

    .. code-block:: python3

        w = Wrapper()

        async def blocking_call():
            with w:
                await asyncio.sleep(10)

        # and somewhere else:
        w.cancel(NoNeedToWaitError('With explanation'))

    """
    _error: Optional[Exception] = None

    cancelled: Optional[bool] = None
    cancel_failed: Optional[bool] = None

    def __init__(self) -> None:
        self._tasks: Set['asyncio.Task[Any]'] = set()

    def __enter__(self) -> None:
        if self._error is not None:
            raise self._error

        task = _current_task()
        if task is None:
            raise RuntimeError('Called not inside a task')

        self._tasks.add(task)

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        task = _current_task()
        assert task
        self._tasks.discard(task)
        if self._error is not None:
            self.cancel_failed = exc_type is not asyncio.CancelledError
            raise self._error

    def cancel(self, error: Exception) -> None:
        self._error = error
        for task in self._tasks:
            task.cancel()
        self.cancelled = True class Status(enum.Enum):
    """Predefined gRPC status codes represented as enum

    See also: https://github.com/grpc/grpc/blob/master/doc/statuscodes.md
    """
    #: The operation completed successfully
    OK = 0
    #: The operation was cancelled (typically by the caller)
    CANCELLED = 1
    #: Generic status to describe error when it can't be described using
    #: other statuses
    UNKNOWN = 2
    #: Client specified an invalid argument
    INVALID_ARGUMENT = 3
    #: Deadline expired before operation could complete
    DEADLINE_EXCEEDED = 4
    #: Some requested entity was not found
    NOT_FOUND = 5
    #: Some entity that we attempted to create already exists
    ALREADY_EXISTS = 6
    #: The caller does not have permission to execute the specified operation
    PERMISSION_DENIED = 7
    #: Some resource has been exhausted, perhaps a per-user quota, or perhaps
    #: the entire file system is out of space
    RESOURCE_EXHAUSTED = 8
    #: Operation was rejected because the system is not in a state required
    #: for the operation's execution
    FAILED_PRECONDITION = 9
    #: The operation was aborted
    ABORTED = 10
    #: Operation was attempted past the valid range
    OUT_OF_RANGE = 11
    #: Operation is not implemented or not supported/enabled in this service
    UNIMPLEMENTED = 12
    #: Internal errors
    INTERNAL = 13
    #: The service is currently unavailable
    UNAVAILABLE = 14
    #: Unrecoverable data loss or corruption
    DATA_LOSS = 15
    #: The request does not have valid authentication credentials for the
    #: operation
    UNAUTHENTICATED = 16 class Cardinality(_Cardinality, enum.Enum):
    UNARY_UNARY = _Cardinality(False, False)
    UNARY_STREAM = _Cardinality(False, True)
    STREAM_UNARY = _Cardinality(True, False)
    STREAM_STREAM = _Cardinality(True, True) class Configuration:
    _keepalive_time: Optional[float] = field(
        default=cast(None, _DEFAULT),
        metadata={
            'validate': _optional(_chain(_of_type(int, float), _positive)),
            'server-default': 7200.0,
            'client-default': None,
            'test-default': None,
        },
    )
    _keepalive_timeout: float = field(
        default=20.0,
        metadata={
            'validate': _chain(_of_type(int, float), _positive),
        },
    )
    _keepalive_permit_without_calls: bool = field(
        default=False,
        metadata={
            'validate': _optional(_of_type(bool)),
        },
    )
    _http2_max_pings_without_data: int = field(
        default=2,
        metadata={
            'validate': _optional(_chain(_of_type(int), _non_negative)),
        },
    )
    _http2_min_sent_ping_interval_without_data: float = field(
        default=300,
        metadata={
            'validate': _optional(_chain(_of_type(int, float), _positive)),
        },
    )
    #: Sets inbound window size for a connection. HTTP/2 spec allows this value
    #: to be from 64 KiB to 2 GiB, 4 MiB is used by default
    http2_connection_window_size: int = field(
        default=_4MiB,
        metadata={
            'validate': _chain(_of_type(int), _range(_WMIN, _WMAX)),
        },
    )
    #: Sets inbound window size for a stream. HTTP/2 spec allows this value
    #: to be from 64 KiB to 2 GiB, 4 MiB is used by default
    http2_stream_window_size: int = field(
        default=_4MiB,
        metadata={
            'validate': _chain(_of_type(int), _range(_WMIN, _WMAX)),
        },
    )

    def __post_init__(self) -> None:
        _validate(self)

    def __for_server__(self) -> 'Configuration':
        return _with_defaults(self, 'server-default')

    def __for_client__(self) -> 'Configuration':
        return _with_defaults(self, 'client-default')

    def __for_test__(self) -> 'Configuration':
        return _with_defaults(self, 'test-default') async def send_message(
    stream: 'Stream',
    codec: 'CodecBase',
    message: _SendType,
    message_type: Type[_SendType],
    *,
    end: bool = False,
) -> None:
    reply_bin = codec.encode(message, message_type)
    reply_data = (struct.pack('?', False)
                  + struct.pack('>I', len(reply_bin))
                  + reply_bin)
    await stream.send_data(reply_data, end_stream=end) async def recv_message(
    stream: 'Stream',
    codec: 'CodecBase',
    message_type: Type[_RecvType],
) -> Optional[_RecvType]:
    meta = await stream.recv_data(5)
    if not meta:
        return None

    compressed_flag = struct.unpack('?', meta[:1])[0]
    if compressed_flag:
        raise NotImplementedError('Compression not implemented')

    message_len = struct.unpack('>I', meta[1:])[0]
    message_bin = await stream.recv_data(message_len)
    assert len(message_bin) == message_len, \
        '{} != {}'.format(len(message_bin), message_len)
    message = codec.decode(message_bin, message_type)
    return cast(_RecvType, message) class StreamIterator(AsyncIterator[_RecvType], metaclass=abc.ABCMeta):

    @abc.abstractmethod
    async def recv_message(self) -> Optional[_RecvType]:
        pass

    def __aiter__(self) -> AsyncIterator[_RecvType]:
        return self

    async def __anext__(self) -> _RecvType:
        message = await self.recv_message()
        if message is None:
            raise StopAsyncIteration()
        else:
            return message async def recv_message(
    stream: 'Stream',
    codec: 'CodecBase',
    message_type: Type[_RecvType],
) -> Optional[_RecvType]:
async def send_message(
    stream: 'Stream',
    codec: 'CodecBase',
    message: _SendType,
    message_type: Type[_SendType],
    *,
    end: bool = False,
) -> None:
    async def recv_message(self) -> Optional[_RecvType]:
    def __aiter__(self) -> AsyncIterator[_RecvType]:
    async def __anext__(self) -> _RecvType:
class StreamIterator(AsyncIterator[_RecvType], metaclass=abc.ABCMeta): class _DispatchServerEvents(_DispatchCommonEvents):

    @_dispatches(RecvRequest)
    async def recv_request(
        self,
        metadata: _Metadata,
        method_func: 'IServerMethodFunc',
        *,
        method_name: str,
        deadline: Optional[Deadline],
        content_type: str,
        user_agent: Optional[str],
        peer: 'Peer',
    ) -> Tuple[_Metadata, 'IServerMethodFunc']:
        return await self.__dispatch__(RecvRequest(  # type: ignore
            metadata=metadata,
            method_func=method_func,
            method_name=method_name,
            deadline=deadline,
            content_type=content_type,
            user_agent=user_agent,
            peer=peer,
        ))

    @_dispatches(SendInitialMetadata)
    async def send_initial_metadata(
        self,
        metadata: _Metadata,
    ) -> Tuple[_Metadata]:
        return await self.__dispatch__(SendInitialMetadata(  # type: ignore
            metadata=metadata,
        ))

    @_dispatches(SendTrailingMetadata)
    async def send_trailing_metadata(
        self,
        metadata: _Metadata,
        *,
        status: Status,
        status_message: Optional[str],
        status_details: Any,
    ) -> Tuple[_Metadata]:
        return await self.__dispatch__(SendTrailingMetadata(  # type: ignore
            metadata=metadata,
            status=status,
            status_message=status_message,
            status_details=status_details,
        )) USER_AGENT = (
    'grpc-python-grpclib/{lib_ver} ({sys}; {py}/{py_ver})'
    .format(
        lib_ver=__version__,
        sys=platform.system(),
        py=platform.python_implementation(),
        py_ver=platform.python_version(),
    )
    .lower()
)
_UNITS = {
    'H': 60 * 60,
    'M': 60,
    'S': 1,
    'm': 10 ** -3,
    'u': 10 ** -6,
    'n': 10 ** -9,
}
_TIMEOUT_RE = re.compile(r'^(\d+)([{}])$'.format(''.join(_UNITS)))
_STATUS_DETAILS_KEY = 'grpc-status-details-bin'
_UNQUOTED = ''.join([chr(i) for i in range(0x20, 0x24 + 1)]
                    + [chr(i) for i in range(0x26, 0x7E + 1)])
_KEY_RE = re.compile(r'^[0-9a-z_.\-]+$')
_VALUE_RE = re.compile(r'^[ !-~]+$')  # 0x20-0x7E - space and printable ASCII
_SPECIAL = {
    'te',
    'content-type',
    'user-agent',
}
def decode_timeout(value: str) -> float:
def encode_timeout(timeout: float) -> str:
    def __init__(self, *, _timestamp: float) -> None:
    def __lt__(self, other: object) -> bool:
    def __eq__(self, other: object) -> bool:
    def from_headers(cls, headers: _Headers) -> Optional['Deadline']:
    def from_timeout(cls, timeout: float) -> 'Deadline':
    def time_remaining(self) -> float:
def encode_grpc_message(message: str) -> str:
def decode_grpc_message(value: str) -> str:
def decode_bin_value(value: bytes) -> bytes:
def decode_metadata(headers: _Headers) -> _Metadata:
def encode_bin_value(value: bytes) -> bytes:
def encode_metadata(metadata: _MetadataLike) -> _Headers:
class Deadline: USER_AGENT = (
    'grpc-python-grpclib/{lib_ver} ({sys}; {py}/{py_ver})'
    .format(
        lib_ver=__version__,
        sys=platform.system(),
        py=platform.python_implementation(),
        py_ver=platform.python_version(),
    )
    .lower()
)
_UNITS = {
    'H': 60 * 60,
    'M': 60,
    'S': 1,
    'm': 10 ** -3,
    'u': 10 ** -6,
    'n': 10 ** -9,
}
_TIMEOUT_RE = re.compile(r'^(\d+)([{}])$'.format(''.join(_UNITS)))
_STATUS_DETAILS_KEY = 'grpc-status-details-bin'
_UNQUOTED = ''.join([chr(i) for i in range(0x20, 0x24 + 1)]
                    + [chr(i) for i in range(0x26, 0x7E + 1)])
_KEY_RE = re.compile(r'^[0-9a-z_.\-]+$')
_VALUE_RE = re.compile(r'^[ !-~]+$')  # 0x20-0x7E - space and printable ASCII
_SPECIAL = {
    'te',
    'content-type',
    'user-agent',
}
def decode_timeout(value: str) -> float:
def encode_timeout(timeout: float) -> str:
    def __init__(self, *, _timestamp: float) -> None:
    def __lt__(self, other: object) -> bool:
    def __eq__(self, other: object) -> bool:
    def from_headers(cls, headers: _Headers) -> Optional['Deadline']:
    def from_timeout(cls, timeout: float) -> 'Deadline':
    def time_remaining(self) -> float:
def encode_grpc_message(message: str) -> str:
def decode_grpc_message(value: str) -> str:
def decode_bin_value(value: bytes) -> bytes:
def decode_metadata(headers: _Headers) -> _Metadata:
def encode_bin_value(value: bytes) -> bytes:
def encode_metadata(metadata: _MetadataLike) -> _Headers:
class Deadline: _STATUS_DETAILS_KEY = 'grpc-status-details-bin' def encode_bin_value(value: bytes) -> bytes:
    return b64encode(value).rstrip(b'=') class H2Protocol(Protocol):
    connection: Connection
    processor: EventsProcessor

    def __init__(
        self,
        handler: AbstractHandler,
        config: Configuration,
        h2_config: H2Configuration,
    ) -> None:
        self.handler = handler
        self.config = config
        self.h2_config = h2_config

    def connection_made(self, transport: BaseTransport) -> None:
        sock = transport.get_extra_info('socket')
        if sock is not None:
            _set_nodelay(sock)

        h2_conn = H2Connection(config=self.h2_config)
        h2_conn.initiate_connection()

        initial = h2_conn.local_settings.initial_window_size
        conn_delta = self.config.http2_connection_window_size - initial
        stream_delta = self.config.http2_stream_window_size - initial
        if conn_delta:
            h2_conn.increment_flow_control_window(conn_delta)
        if stream_delta:
            h2_conn.update_settings({
                SettingCodes.INITIAL_WINDOW_SIZE:
                    self.config.http2_stream_window_size,
            })

        self.connection = Connection(
            h2_conn,
            cast(Transport, transport),
            config=self.config,
        )
        self.connection.flush()
        self.connection.initialize()

        self.processor = EventsProcessor(self.handler, self.connection)

    def data_received(self, data: bytes) -> None:
        try:
            events = self.connection.feed(data)
        except ProtocolError:
            log.debug('Protocol error', exc_info=True)
            self.processor.close('Protocol error')
        else:
            self.connection.flush()
            for event in events:
                self.processor.process(event)
            self.connection.flush()

    def pause_writing(self) -> None:
        self.connection.pause_writing()

    def resume_writing(self) -> None:
        self.connection.resume_writing()

    def connection_lost(self, exc: Optional[BaseException]) -> None:
        self.processor.close(reason='Connection lost') class AbstractHandler(ABC):

    @abstractmethod
    def accept(
        self,
        stream: Stream,
        headers: _Headers,
        release_stream: Callable[[], None],
    ) -> None:
        pass

    @abstractmethod
    def cancel(self, stream: Stream) -> None:
        pass

    @abstractmethod
    def close(self) -> None:
        pass class GRPCError(Exception):
    """Expected error, may be raised during RPC call

    There can be multiple origins of this error. It can be generated
    on the server-side and on the client-side. If this error originates from
    the server, on the wire this error is represented as ``grpc-status`` and
    ``grpc-message`` trailers. Possible values of the ``grpc-status`` trailer
    are described in the gRPC protocol definition. In ``grpclib`` these values
    are represented as :py:class:`~grpclib.const.Status` enum.

    Here are possible origins of this error:

      - you may raise this error to cancel current call on the server-side or
        return non-OK :py:class:`~grpclib.const.Status` using
        :py:meth:`~grpclib.server.Stream.send_trailing_metadata` method
        `(e.g. resource not found)`
      - server may return non-OK ``grpc-status`` in different failure
        conditions `(e.g. invalid request)`
      - client raises this error for non-OK ``grpc-status`` from the server
      - client may raise this error in different failure conditions
        `(e.g. server returned unsupported` ``:content-type`` `header)`

    """
    def __init__(
        self,
        status: Status,
        message: Optional[str] = None,
        details: Any = None,
    ) -> None:
        super().__init__(status, message, details)
        #: :py:class:`~grpclib.const.Status` of the error
        self.status = status
        #: Error message
        self.message = message
        #: Error details
        self.details = details class ProtocolError(Exception):
    """Unexpected error, raised by ``grpclib`` when your code violates
    gRPC protocol

    This error means that you probably should fix your code.
    """ class StreamTerminatedError(Exception):
    """Unexpected error, raised when we receive ``RST_STREAM`` frame from
    the other side

    This error means that the other side decided to forcefully cancel current
    call, probably because of a protocol error.
    """ GRPC_CONTENT_TYPE = 'application/grpc' class CodecBase(abc.ABC):

    @property
    @abc.abstractmethod
    def __content_subtype__(self) -> str:
        pass

    @abc.abstractmethod
    def encode(self, message: Any, message_type: Any) -> bytes:
        pass

    @abc.abstractmethod
    def decode(self, data: bytes, message_type: Any) -> Any:
        pass class StatusDetailsCodecBase(abc.ABC):

    @abc.abstractmethod
    def encode(
        self, status: Status, message: Optional[str], details: Any,
    ) -> bytes:
        pass

    @abc.abstractmethod
    def decode(
        self, status: Status, message: Optional[str], data: bytes,
    ) -> Any:
        pass class ProtoCodec(CodecBase):
    __content_subtype__ = 'proto'

    def encode(
        self,
        message: 'IProtoMessage',
        message_type: Type['IProtoMessage'],
    ) -> bytes:
        if not isinstance(message, message_type):
            raise TypeError('Message must be of type {!r}, not {!r}'
                            .format(message_type, type(message)))
        return message.SerializeToString()

    def decode(
        self,
        data: bytes,
        message_type: Type['IProtoMessage'],
    ) -> 'IProtoMessage':
        return message_type.FromString(data) class ProtoStatusDetailsCodec(StatusDetailsCodecBase):

    def encode(
        self,
        status: Status,
        message: Optional[str],
        details: Sequence['Message'],
    ) -> bytes:
        status_pb2 = _status_pb2()

        status_proto = status_pb2.Status(code=status.value, message=message)
        if details is not None:
            for detail in details:
                detail_container = status_proto.details.add()
                detail_container.Pack(detail)
        return status_proto.SerializeToString()  # type: ignore

    def decode(
        self, status: Status, message: Optional[str], data: bytes,
    ) -> Sequence[Any]:
        status_pb2 = _status_pb2()
        sym_db = _sym_db()

        status_proto = status_pb2.Status.FromString(data)
        details = []
        for detail_container in status_proto.details:
            try:
                msg_type = sym_db.GetSymbol(detail_container.TypeName())
            except KeyError:
                details.append(_Unknown(detail_container.TypeName()))
                continue
            detail = msg_type()
            detail_container.Unpack(detail)
            details.append(detail)
        return details @_cached
def _googleapis_available() -> bool:
    try:
        import google.rpc.status_pb2  # noqa
    except ImportError:
        return False
    else:
        return True
import abc
import time
import socket
import logging
import asyncio
import warnings
import h2.config
import h2.exceptions
    import ssl as _ssl  # noqa
from types import TracebackType
from typing import TYPE_CHECKING, Optional, Collection, Generic, Type, cast
from typing import List, Tuple, Dict, Any, Callable, ContextManager, Set
from contextlib import nullcontext
from multidict import MultiDict
from .utils import DeadlineWrapper, Wrapper
from .const import Status, Cardinality
from .config import Configuration
from .stream import send_message, recv_message, StreamIterator
from .stream import _RecvType, _SendType
from .events import _DispatchServerEvents
from .metadata import Deadline, encode_grpc_message, _Metadata
from .metadata import encode_metadata, decode_metadata, _MetadataLike
from .metadata import _STATUS_DETAILS_KEY, encode_bin_value
from .protocol import H2Protocol, AbstractHandler
from .exceptions import GRPCError, ProtocolError, StreamTerminatedError
from .encoding.base import GRPC_CONTENT_TYPE, CodecBase, StatusDetailsCodecBase
from .encoding.proto import ProtoCodec, ProtoStatusDetailsCodec
from .encoding.proto import _googleapis_available
from ._registry import servers as _servers
    from . import const  # noqa
    from . import protocol  # noqa
    from ._typing import IServable  # noqa
        stream: 'protocol.Stream',
        method_name: str,
        cardinality: Cardinality,
        recv_type: Type[_RecvType],
        send_type: Type[_SendType],
        *,
        codec: CodecBase,
        status_details_codec: Optional[StatusDetailsCodecBase],
        dispatch: _DispatchServerEvents,
        deadline: Optional[Deadline] = None,
        user_agent: Optional[str] = None,
    ):
        self._stream = stream
        self._method_name = method_name
        self._cardinality = cardinality
        self._recv_type = recv_type
        self._send_type = send_type
        self._codec = codec
        self._status_details_codec = status_details_codec
        self._dispatch = dispatch
        #: :py:class:`~grpclib.metadata.Deadline` of the current request
        self.deadline = deadline
        #: Invocation metadata, received with headers from the client.
        #: Represented as a multi-dict object.
        self.metadata: Optional[_Metadata] = None
        #: Client's user-agent
        self.user_agent = user_agent
        #: Connection's peer info of type :py:class:`~grpclib.protocol.Peer`
        self.peer = self._stream.connection.get_peer()

    @property
    def _content_type(self) -> str:
        return GRPC_CONTENT_TYPE + '+' + self._codec.__content_subtype__

    async def recv_message(self) -> Optional[_RecvType]:
        """Coroutine to receive incoming message from the client.

        If client sends UNARY request, then you can call this coroutine
        only once. If client sends STREAM request, then you should call this
        coroutine several times, until it returns None when the client has
        ended the stream. To simplify your code in this case,
        :py:class:`Stream` class implements async iteration protocol, so
        you can use it like this:

        .. code-block:: python3

            async for message in stream:
                do_smth_with(message)

        or even like this:

        .. code-block:: python3

            messages = [msg async for msg in stream]

        HTTP/2 has flow control mechanism, so server will acknowledge received
        DATA frames as a message only after user consumes this coroutine.

        :returns: message
        """
        message = await recv_message(self._stream, self._codec, self._recv_type)
        if message is not None:
            message, = await self._dispatch.recv_message(message)
            self._messages_received += 1
            self._stream.connection.messages_received += 1
            self._stream.connection.last_message_received = time.monotonic()
            return message
        else:
            return None

    async def send_initial_metadata(
        self,
        *,
        metadata: Optional[_MetadataLike] = None,
    ) -> None:
        """Coroutine to send headers with initial metadata to the client.

        In gRPC you can send initial metadata as soon as possible, because
        gRPC doesn't use `:status` pseudo header to indicate success or failure
        of the current request. gRPC uses trailers for this purpose, and
        trailers are sent during :py:meth:`send_trailing_metadata` call, which
        should be called in the end.

        .. note:: This coroutine will be called implicitly during first
            :py:meth:`send_message` coroutine call, if not called before
            explicitly.

        :param metadata: custom initial metadata, dict or list of pairs
        """
        if self._send_initial_metadata_done:
            raise ProtocolError('Initial metadata was already sent')

        headers = [
            (':status', '200'),
            ('content-type', self._content_type),
        ]
        metadata = MultiDict(metadata or ())
        metadata, = await self._dispatch.send_initial_metadata(metadata)
        headers.extend(encode_metadata(cast(_Metadata, metadata)))

        await self._stream.send_headers(headers)
        self._send_initial_metadata_done = True

    async def send_message(self, message: _SendType) -> None:
        """Coroutine to send message to the client.

        If server sends UNARY response, then you should call this coroutine only
        once. If server sends STREAM response, then you can call this coroutine
        as many times as you need.

        :param message: message object
        """
        if not self._send_initial_metadata_done:
            await self.send_initial_metadata()

        if not self._cardinality.server_streaming:
            if self._send_message_done:
                raise ProtocolError('Message was already sent')

        message, = await self._dispatch.send_message(message)

        await send_message(self._stream, self._codec, message, self._send_type)