|
| 1 | +# SPDX-FileCopyrightText: Copyright (c) 2022 Kevin Conley |
| 2 | +# SPDX-License-Identifier: MIT |
| 3 | +""" |
| 4 | +`circuitpython_typing.socket` |
| 5 | +================================================================================ |
| 6 | +
|
| 7 | +Type annotation definitions for sockets. Used for `adafruit_requests` and similar libraries. |
| 8 | +
|
| 9 | +* Author(s): Kevin Conley |
| 10 | +""" |
| 11 | + |
| 12 | +from ssl import SSLContext |
| 13 | +from types import ModuleType |
| 14 | +from typing import Any, Optional, Protocol, Tuple, Union |
| 15 | + |
| 16 | +# Based on https://github.com/python/typeshed/blob/master/stdlib/_socket.pyi |
| 17 | + |
| 18 | + |
| 19 | +class CommonSocketType(Protocol): |
| 20 | + """Describes the common structure every socket type must have.""" |
| 21 | + |
| 22 | + def send(self, data: bytes, flags: int = ...) -> None: |
| 23 | + """Send data to the socket. The meaning of the optional flags kwarg is |
| 24 | + implementation-specific.""" |
| 25 | + ... |
| 26 | + |
| 27 | + def settimeout(self, value: Optional[float]) -> None: |
| 28 | + """Set a timeout on blocking socket operations.""" |
| 29 | + ... |
| 30 | + |
| 31 | + def close(self) -> None: |
| 32 | + """Close the socket.""" |
| 33 | + ... |
| 34 | + |
| 35 | + |
| 36 | +class CommonCircuitPythonSocketType(CommonSocketType, Protocol): |
| 37 | + """Describes the common structure every CircuitPython socket type must have.""" |
| 38 | + |
| 39 | + def connect( |
| 40 | + self, |
| 41 | + address: Tuple[str, int], |
| 42 | + conntype: Optional[int] = ..., |
| 43 | + ) -> None: |
| 44 | + """Connect to a remote socket at the provided (host, port) address. The conntype |
| 45 | + kwarg optionally may indicate SSL or not, depending on the underlying interface.""" |
| 46 | + ... |
| 47 | + |
| 48 | + |
| 49 | +class LegacyCircuitPythonSocketType(CommonCircuitPythonSocketType, Protocol): |
| 50 | + """Describes the structure a legacy CircuitPython socket type must have.""" |
| 51 | + |
| 52 | + def recv(self, bufsize: int = ...) -> bytes: |
| 53 | + """Receive data from the socket. The return value is a bytes object representing |
| 54 | + the data received. The maximum amount of data to be received at once is specified |
| 55 | + by bufsize.""" |
| 56 | + ... |
| 57 | + |
| 58 | + |
| 59 | +class SupportsRecvWithFlags(Protocol): |
| 60 | + """Describes a type that posseses a socket recv() method supporting the flags kwarg.""" |
| 61 | + |
| 62 | + def recv(self, bufsize: int = ..., flags: int = ...) -> bytes: |
| 63 | + """Receive data from the socket. The return value is a bytes object representing |
| 64 | + the data received. The maximum amount of data to be received at once is specified |
| 65 | + by bufsize. The meaning of the optional flags kwarg is implementation-specific.""" |
| 66 | + ... |
| 67 | + |
| 68 | + |
| 69 | +class SupportsRecvInto(Protocol): |
| 70 | + """Describes a type that possesses a socket recv_into() method.""" |
| 71 | + |
| 72 | + def recv_into(self, buffer: bytearray, nbytes: int = ..., flags: int = ...) -> int: |
| 73 | + """Receive up to nbytes bytes from the socket, storing the data into the provided |
| 74 | + buffer. If nbytes is not specified (or 0), receive up to the size available in the |
| 75 | + given buffer. The meaning of the optional flags kwarg is implementation-specific. |
| 76 | + Returns the number of bytes received.""" |
| 77 | + ... |
| 78 | + |
| 79 | + |
| 80 | +class CircuitPythonSocketType( |
| 81 | + CommonCircuitPythonSocketType, |
| 82 | + SupportsRecvInto, |
| 83 | + SupportsRecvWithFlags, |
| 84 | + Protocol, |
| 85 | +): # pylint: disable=too-many-ancestors |
| 86 | + """Describes the structure every modern CircuitPython socket type must have.""" |
| 87 | + |
| 88 | + ... |
| 89 | + |
| 90 | + |
| 91 | +class StandardPythonSocketType( |
| 92 | + CommonSocketType, SupportsRecvInto, SupportsRecvWithFlags, Protocol |
| 93 | +): |
| 94 | + """Describes the structure every standard Python socket type must have.""" |
| 95 | + |
| 96 | + def connect(self, address: Union[Tuple[Any, ...], str, bytes]) -> None: |
| 97 | + """Connect to a remote socket at the provided address.""" |
| 98 | + ... |
| 99 | + |
| 100 | + |
| 101 | +SocketType = Union[ |
| 102 | + LegacyCircuitPythonSocketType, |
| 103 | + CircuitPythonSocketType, |
| 104 | + StandardPythonSocketType, |
| 105 | +] |
| 106 | + |
| 107 | +SocketpoolModuleType = ModuleType |
| 108 | + |
| 109 | + |
| 110 | +class InterfaceType(Protocol): |
| 111 | + """Describes the structure every interface type must have.""" |
| 112 | + |
| 113 | + @property |
| 114 | + def TLS_MODE(self) -> int: # pylint: disable=invalid-name |
| 115 | + """Constant representing that a socket's connection mode is TLS.""" |
| 116 | + ... |
| 117 | + |
| 118 | + |
| 119 | +SSLContextType = Union[SSLContext, "_FakeSSLContext"] |
0 commit comments