3737__repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_Requests.git"
3838
3939import errno
40+ import sys
41+
42+ if sys .implementation .name == "circuitpython" :
43+
44+ def cast (_t , value ):
45+ """No-op shim for the typing.cast() function which is not available in CircuitPython."""
46+ return value
47+
48+
49+ else :
50+ from ssl import SSLContext
51+ from types import ModuleType , TracebackType
52+ from typing import Any , Dict , List , Optional , Protocol , Tuple , Type , Union , cast
53+
54+ # Based on https://github.com/python/typeshed/blob/master/stdlib/_socket.pyi
55+ class CommonSocketType (Protocol ):
56+ """Describes the common structure every socket type must have."""
57+
58+ def send (self , data : bytes , flags : int = ...) -> None :
59+ """Send data to the socket. The meaning of the optional flags kwarg is
60+ implementation-specific."""
61+ ...
62+
63+ def settimeout (self , value : Optional [float ]) -> None :
64+ """Set a timeout on blocking socket operations."""
65+ ...
66+
67+ def close (self ) -> None :
68+ """Close the socket."""
69+ ...
70+
71+ class CommonCircuitPythonSocketType (CommonSocketType , Protocol ):
72+ """Describes the common structure every CircuitPython socket type must have."""
73+
74+ def connect (
75+ self ,
76+ address : Tuple [str , int ],
77+ conntype : Optional [int ] = ...,
78+ ) -> None :
79+ """Connect to a remote socket at the provided (host, port) address. The conntype
80+ kwarg optionally may indicate SSL or not, depending on the underlying interface."""
81+ ...
82+
83+ class LegacyCircuitPythonSocketType (CommonCircuitPythonSocketType , Protocol ):
84+ """Describes the structure a legacy CircuitPython socket type must have."""
85+
86+ def recv (self , bufsize : int = ...) -> bytes :
87+ """Receive data from the socket. The return value is a bytes object representing
88+ the data received. The maximum amount of data to be received at once is specified
89+ by bufsize."""
90+ ...
91+
92+ class SupportsRecvWithFlags (Protocol ):
93+ """Describes a type that posseses a socket recv() method supporting the flags kwarg."""
94+
95+ def recv (self , bufsize : int = ..., flags : int = ...) -> bytes :
96+ """Receive data from the socket. The return value is a bytes object representing
97+ the data received. The maximum amount of data to be received at once is specified
98+ by bufsize. The meaning of the optional flags kwarg is implementation-specific."""
99+ ...
100+
101+ class SupportsRecvInto (Protocol ):
102+ """Describes a type that possesses a socket recv_into() method."""
103+
104+ def recv_into (
105+ self , buffer : bytearray , nbytes : int = ..., flags : int = ...
106+ ) -> int :
107+ """Receive up to nbytes bytes from the socket, storing the data into the provided
108+ buffer. If nbytes is not specified (or 0), receive up to the size available in the
109+ given buffer. The meaning of the optional flags kwarg is implementation-specific.
110+ Returns the number of bytes received."""
111+ ...
112+
113+ class CircuitPythonSocketType (
114+ CommonCircuitPythonSocketType ,
115+ SupportsRecvInto ,
116+ SupportsRecvWithFlags ,
117+ Protocol ,
118+ ): # pylint: disable=too-many-ancestors
119+ """Describes the structure every modern CircuitPython socket type must have."""
120+
121+ ...
122+
123+ class StandardPythonSocketType (
124+ CommonSocketType , SupportsRecvInto , SupportsRecvWithFlags , Protocol
125+ ):
126+ """Describes the structure every standard Python socket type must have."""
40127
41- try :
42- from typing import Union , TypeVar , Optional , Dict , Any , List , Type
43- import types
44- from types import TracebackType
45- import ssl
46- import adafruit_esp32spi .adafruit_esp32spi_socket as esp32_socket
47- import adafruit_wiznet5k .adafruit_wiznet5k_socket as wiznet_socket
48- import adafruit_fona .adafruit_fona_socket as cellular_socket
49- from adafruit_esp32spi .adafruit_esp32spi import ESP_SPIcontrol
50- from adafruit_wiznet5k .adafruit_wiznet5k import WIZNET5K
51- from adafruit_fona .adafruit_fona import FONA
52- import socket as cpython_socket
53-
54- SocketType = TypeVar (
55- "SocketType" ,
56- esp32_socket .socket ,
57- wiznet_socket .socket ,
58- cellular_socket .socket ,
59- cpython_socket .socket ,
60- )
61- SocketpoolModuleType = types .ModuleType
62- SSLContextType = (
63- ssl .SSLContext
64- ) # Can use either CircuitPython or CPython ssl module
65- InterfaceType = TypeVar ("InterfaceType" , ESP_SPIcontrol , WIZNET5K , FONA )
128+ def connect (self , address : Union [Tuple [Any , ...], str , bytes ]) -> None :
129+ """Connect to a remote socket at the provided address."""
130+ ...
131+
132+ SocketType = Union [
133+ LegacyCircuitPythonSocketType ,
134+ CircuitPythonSocketType ,
135+ StandardPythonSocketType ,
136+ ]
137+
138+ SocketpoolModuleType = ModuleType
139+
140+ class InterfaceType (Protocol ):
141+ """Describes the structure every interface type must have."""
142+
143+ @property
144+ def TLS_MODE (self ) -> int : # pylint: disable=invalid-name
145+ """Constant representing that a socket's connection mode is TLS."""
146+ ...
147+
148+ SSLContextType = Union [SSLContext , "_FakeSSLContext" ]
66149
67- except ImportError :
68- pass
69150
70151# CircuitPython 6.0 does not have the bytearray.split method.
71152# This function emulates buf.split(needle)[0], which is the functionality
@@ -157,7 +238,7 @@ def _recv_into(self, buf: bytearray, size: int = 0) -> int:
157238 read_size = len (b )
158239 buf [:read_size ] = b
159240 return read_size
160- return self .socket .recv_into (buf , size )
241+ return cast ( "SupportsRecvInto" , self .socket ) .recv_into (buf , size )
161242
162243 @staticmethod
163244 def _find (buf : bytes , needle : bytes , start : int , end : int ) -> int :
@@ -440,7 +521,7 @@ def _free_sockets(self) -> None:
440521
441522 def _get_socket (
442523 self , host : str , port : int , proto : str , * , timeout : float = 1
443- ) -> SocketType :
524+ ) -> CircuitPythonSocketType :
444525 # pylint: disable=too-many-branches
445526 key = (host , port , proto )
446527 if key in self ._open_sockets :
@@ -693,15 +774,15 @@ def delete(self, url: str, **kw) -> Response:
693774
694775
695776class _FakeSSLSocket :
696- def __init__ (self , socket : SocketType , tls_mode : int ) -> None :
777+ def __init__ (self , socket : CircuitPythonSocketType , tls_mode : int ) -> None :
697778 self ._socket = socket
698779 self ._mode = tls_mode
699780 self .settimeout = socket .settimeout
700781 self .send = socket .send
701782 self .recv = socket .recv
702783 self .close = socket .close
703784
704- def connect (self , address : Union [ bytes , str ]) -> None :
785+ def connect (self , address : Tuple [ str , int ]) -> None :
705786 """connect wrapper to add non-standard mode parameter"""
706787 try :
707788 return self ._socket .connect (address , self ._mode )
@@ -714,7 +795,7 @@ def __init__(self, iface: InterfaceType) -> None:
714795 self ._iface = iface
715796
716797 def wrap_socket (
717- self , socket : SocketType , server_hostname : Optional [str ] = None
798+ self , socket : CircuitPythonSocketType , server_hostname : Optional [str ] = None
718799 ) -> _FakeSSLSocket :
719800 """Return the same socket"""
720801 # pylint: disable=unused-argument
0 commit comments