"""
- _instance_cache = None # type: Dict[IOLoop, AsyncHTTPClient]
+ _instance_cache: Dict[IOLoop, "AsyncHTTPClient"]
@classmethod
def configurable_base(cls) -> Type[Configurable]:
class HTTPRequest:
"""HTTP client request object."""
- _headers = None # type: Union[Dict[str, str], httputil.HTTPHeaders]
+ _headers: Union[Dict[str, str], httputil.HTTPHeaders]
# Default values for HTTPRequest parameters.
# Merged with the values on the request object by AsyncHTTPClient
# I'm not sure why these don't get type-inferred from the references in __init__.
error = None # type: Optional[BaseException]
_error_is_response_code = False
- request = None # type: HTTPRequest
+ request: HTTPRequest
def __init__(
self,
temporarily restored in 6.5.2.
"""
- path = None # type: str
- query = None # type: str
+ path: str
+ query: str
# HACK: Used for stream_request_body
- _body_future = None # type: Future[None]
+ _body_future: "Future[None]"
def __init__(
self,
wrapped when `IOStream.connect` is finished.
"""
- socket = None # type: ssl.SSLSocket
+ socket: ssl.SSLSocket
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""The ``ssl_options`` keyword argument may either be an
of this class.
"""
- _threadpool = None # type: ignore
- _threadpool_pid = None # type: int
+ _threadpool: Optional[concurrent.futures.ThreadPoolExecutor] = None
+ _threadpool_pid: Optional[int] = None
def initialize(self, num_threads: int = 10) -> None: # type: ignore
threadpool = ThreadedResolver._create_threadpool(num_threads)
class FinalReturnTest(WebTestCase):
- final_return = None # type: Future
+ final_return: Future
def get_handlers(self):
test = self
# There may be a clever way to use generics here to get more
# precise types (i.e. for a particular Configurable subclass T,
# all the types are subclasses of T, not just Configurable).
- __impl_class = None # type: Optional[Type[Configurable]]
- __impl_kwargs = None # type: Dict[str, Any]
+ __impl_class: Optional[Type["Configurable"]] = None
+ __impl_kwargs: Optional[Dict[str, Any]] = None
def __new__(cls, *args: Any, **kwargs: Any) -> Any:
base = cls.configurable_base()
@classmethod
def _save_configuration(cls):
- # type: () -> Tuple[Optional[Type[Configurable]], Dict[str, Any]]
+ # type: () -> Tuple[Optional[Type[Configurable]], Optional[Dict[str, Any]]]
base = cls.configurable_base()
return (base.__impl_class, base.__impl_kwargs)
@classmethod
def _restore_configuration(cls, saved):
- # type: (Tuple[Optional[Type[Configurable]], Dict[str, Any]]) -> None
+ # type: (Tuple[Optional[Type[Configurable]], Optional[Dict[str, Any]]]) -> None
base = cls.configurable_base()
base.__impl_class = saved[0]
base.__impl_kwargs = saved[1]
_stream_request_body = False
# Will be set in _execute.
- _transforms = None # type: List[OutputTransform]
- path_args = None # type: List[str]
- path_kwargs = None # type: Dict[str, str]
+ _transforms: List["OutputTransform"]
+ path_args: List[str]
+ path_kwargs: Dict[str, str]
def __init__(
self,
RSV_MASK = RSV1 | RSV2 | RSV3
OPCODE_MASK = 0x0F
- stream = None # type: IOStream
+ stream: IOStream
def __init__(
self,
`websocket_connect` function instead.
"""
- protocol = None # type: WebSocketProtocol
+ protocol: Optional[WebSocketProtocol] = None
def __init__(
self,
.. versionadded:: 5.1
"""
+ assert self.protocol is not None
return self.protocol.selected_subprotocol
def log_exception(