from typing import Any
from collections.abc import Callable
-import typing
-
-if typing.TYPE_CHECKING:
- from typing import Deque, Tuple # noqa: F401
curl_log = logging.getLogger("tornado.curl_httpclient")
self._multi.setopt(pycurl.M_SOCKETFUNCTION, self._handle_socket)
self._curls = [self._curl_create() for i in range(max_clients)]
self._free_list = self._curls[:]
- self._requests: Deque[
+ self._requests: collections.deque[
tuple[HTTPRequest, Callable[[HTTPResponse], None], float]
] = collections.deque()
self._fds: dict[int, int] = {}
Dict,
overload,
)
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
from collections.abc import Mapping, Awaitable, Sequence
-if typing.TYPE_CHECKING:
- from typing import Deque, Optional, Set # noqa: F401
- from collections.abc import Iterable
-
_T = typing.TypeVar("_T")
_Yieldable = Union[
self._unfinished = {f: i for (i, f) in enumerate(args)}
futures = args
- self._finished: Deque[Future] = collections.deque()
+ self._finished: collections.deque[Future] = collections.deque()
self.current_index: str | int | None = None
self.current_future: Future | None = None
self._running_future: Future | None = None
from collections.abc import Callable
from collections.abc import Awaitable
-if typing.TYPE_CHECKING:
- from typing import Set # noqa: F401
-
class HTTPServer(TCPServer, Configurable, httputil.HTTPServerConnectionDelegate):
r"""A non-blocking, single-threaded HTTP server.
from collections.abc import Iterable, Mapping, Iterator, Awaitable, Generator
if typing.TYPE_CHECKING:
- from typing import Deque # noqa: F401
- from asyncio import Future # noqa: F401
- import unittest # noqa: F401
+ # These are relatively heavy imports and aren't needed in this file
+ # unless we're type-checking.
+ from asyncio import Future
+ import unittest
# To be used with str.strip() and related methods.
HTTP_WHITESPACE = " \t"
from tornado.util import Configurable, TimeoutError, import_object
import typing
-from typing import Any, TypeVar
+from typing import Any, TypeVar, TypedDict, Protocol
from collections.abc import Callable
from collections.abc import Awaitable
-if typing.TYPE_CHECKING:
- from typing import Dict, List, Set, TypedDict # noqa: F401
-
- from typing_extensions import Protocol
-else:
- Protocol = object
-
class _Selectable(Protocol):
def fileno(self) -> int:
.. versionchanged:: 6.2
``tornado.util.TimeoutError`` is now an alias to ``asyncio.TimeoutError``.
"""
- if typing.TYPE_CHECKING:
- class FutureCell(TypedDict):
- # noqa: F841
- future: Future | None
- timeout_called: bool
+ class FutureCell(TypedDict):
+ # noqa: F841
+ future: Future | None
+ timeout_called: bool
future_cell: FutureCell = {"future": None, "timeout_called": False}
import typing
from typing import (
- Optional,
Any,
TypeVar,
)
from re import Pattern
from types import TracebackType
-if typing.TYPE_CHECKING:
- from typing import Deque, List, Type # noqa: F401
-
_IOStreamType = TypeVar("_IOStreamType", bound="IOStream")
# These errnos indicate that a connection has been abruptly terminated.
def __init__(self) -> None:
# A sequence of (False, bytearray) and (True, memoryview) objects
- self._buffers: Deque[tuple[bool, bytearray | memoryview]] = collections.deque()
+ self._buffers: collections.deque[tuple[bool, bytearray | memoryview]] = (
+ collections.deque()
+ )
# Position in the first buffer
self._first_pos = 0
self._size = 0
self._read_partial = False
self._read_until_close = False
self._read_future: Future | None = None
- self._write_futures: Deque[tuple[int, Future[None]]] = collections.deque()
+ self._write_futures: collections.deque[tuple[int, Future[None]]] = (
+ collections.deque()
+ )
self._close_callback: Callable[[], None] | None = None
self._connect_future: Future[IOStream] | None = None
# _ssl_connect_future should be defined in SSLIOStream
| bool
| BaseException
| tuple[
- "Optional[Type[BaseException]]",
+ type[BaseException] | None,
BaseException | None,
TracebackType | None,
]
from typing import Optional, Type, Any
from collections.abc import Awaitable
-import typing
-
-if typing.TYPE_CHECKING:
- from typing import Deque, Set # noqa: F401
__all__ = ["Condition", "Event", "Semaphore", "BoundedSemaphore", "Lock"]
"""
def __init__(self) -> None:
- self._waiters: Deque[Future] = collections.deque()
+ self._waiters: collections.deque[Future] = collections.deque()
self._timeouts = 0
def _garbage_collect(self) -> None:
from tornado.netutil import Resolver, is_valid_ip
import typing
-
-if typing.TYPE_CHECKING:
- from typing import Any, List, Tuple, Dict # noqa: F401
- from collections.abc import Generator
+from typing import Any
+from collections.abc import Generator
class CaresResolver(Resolver):
@gen.coroutine
def resolve(
self, host: str, port: int, family: int = 0
- ) -> "Generator[Any, Any, List[Tuple[int, Any]]]":
+ ) -> "Generator[Any, Any, list[tuple[int, Any]]]":
if is_valid_ip(host):
addresses = [host]
else:
from tornado.concurrent import Future, future_set_exc_info
from tornado import gen
-import typing # noqa: F401
+import typing
def install() -> None:
from tornado.iostream import PipeIOStream
from tornado.log import gen_log
-import typing
from typing import Any
from collections.abc import Callable
-if typing.TYPE_CHECKING:
- from typing import List # noqa: F401
-
# Re-export this exception for convenience.
CalledProcessError = subprocess.CalledProcessError
from tornado.concurrent import Future, future_set_result_unless_cancelled
from tornado.locks import Event
-from typing import TypeVar, Generic
+from typing import TypeVar, Generic, Any
from collections.abc import Awaitable
-import typing
-
-if typing.TYPE_CHECKING:
- from typing import Deque, Tuple, Any # noqa: F401
_T = TypeVar("_T")
self._maxsize = maxsize
self._init()
- self._getters: Deque[Future[_T]] = collections.deque([])
- self._putters: Deque[tuple[_T, Future[None]]] = collections.deque([])
+ self._getters: collections.deque[Future[_T]] = collections.deque([])
+ self._putters: collections.deque[tuple[_T, Future[None]]] = collections.deque(
+ []
+ )
self._unfinished_tasks = 0
self._finished = Event()
self._finished.set()
from collections.abc import Callable
from collections.abc import Awaitable
from types import TracebackType
-import typing
-
-if typing.TYPE_CHECKING:
- from typing import Deque, Tuple, List # noqa: F401
class HTTPTimeoutError(HTTPError):
) -> None:
super().initialize(defaults=defaults)
self.max_clients = max_clients
- self.queue: Deque[
+ self.queue: collections.deque[
tuple[object, HTTPRequest, Callable[[HTTPResponse], None]]
] = collections.deque()
self.active: dict[
import numbers
import datetime
import ssl
-import typing
from tornado.concurrent import Future, future_add_done_callback
from tornado.ioloop import IOLoop
from collections.abc import Callable
from collections.abc import Iterator
-if typing.TYPE_CHECKING:
- from typing import Set # noqa(F401)
-
_INITIAL_CONNECT_TIMEOUT = 0.3
from tornado import process
from tornado.util import errno_from_exception
-import typing
from typing import Any
-from collections.abc import Iterable, Awaitable
-
-if typing.TYPE_CHECKING:
- from typing import List # noqa: F401
- from collections.abc import Callable
+from collections.abc import Iterable, Awaitable, Callable
class TCPServer:
from tornado.log import app_log
from tornado.util import ObjectDict, exec_in, unicode_type
-from typing import Any, Optional, TextIO
+from typing import Any, Optional, TextIO, ContextManager
from collections.abc import Callable
from collections.abc import Iterable
import typing
-if typing.TYPE_CHECKING:
- from typing import Tuple, ContextManager # noqa: F401
-
_DEFAULT_AUTOESCAPE = "xhtml_escape"
)
from tornado.util import unicode_type
-from typing import List, Tuple, Union, Dict, Any # noqa: F401
+from typing import Any
linkify_tests: list[tuple[str | bytes, dict[str, Any], str]] = [
# (input, linkify_kwargs, expected_output)
except ImportError:
contextvars = None # type: ignore
-import typing
-
-if typing.TYPE_CHECKING:
- from typing import List, Optional # noqa: F401
-
class GenBasicTest(AsyncTestCase):
@gen.coroutine
import socket
-import typing # noqa(F401)
from tornado.http1connection import HTTP1Connection
from tornado.httputil import HTTPMessageDelegate
import subprocess
import sys
import time
-import typing # noqa: F401
import unicodedata
import unittest
import typing
-if typing.TYPE_CHECKING:
- from typing import Dict, List # noqa: F401
-
async def read_stream_body(stream):
"""Reads an HTTP response from `stream` and returns a tuple of its
)
from tornado.concurrent import Future
-import typing
-
-if typing.TYPE_CHECKING:
- from typing import List # noqa: F401
-
class TestIOLoop(AsyncTestCase):
def test_add_callback_return_sequence(self):
import asyncio
from datetime import timedelta
-import typing # noqa: F401
import unittest
from tornado import gen, locks
from tornado.options import OptionParser, Error
from tornado.util import basestring_type
-import typing
-
-if typing.TYPE_CHECKING:
- from typing import List # noqa: F401
-
class Email:
def __init__(self, value):
from tornado.web import Application, HTTPError, RequestHandler
from tornado.wsgi import WSGIContainer
-import typing # noqa: F401
+import typing
class BasicRouter(Router):
import socket
import ssl
import sys
-import typing # noqa: F401
+import typing
from tornado.escape import to_unicode, utf8
from tornado import gen, version
import unittest
from tornado.concurrent import Future
+from tornado.iostream import IOStream
from tornado.netutil import bind_sockets, Resolver
from tornado.queues import Queue
from tornado.tcpclient import TCPClient, _Connector
import typing
-if typing.TYPE_CHECKING:
- from tornado.iostream import IOStream # noqa: F401
- from typing import List, Dict, Tuple # noqa: F401
-
# Fake address families for testing. Used in place of AF_INET
# and AF_INET6 because some installations do not have AF_INET6.
AF1, AF2 = 1, 2
from tornado.template import Template, DictLoader, ParseError, Loader
from tornado.util import ObjectDict
-import typing # noqa: F401
-
class TemplateTest(unittest.TestCase):
def test_simple(self):
from collections.abc import Coroutine
from types import TracebackType
-if typing.TYPE_CHECKING:
- _ExcInfoTuple = tuple[
- Optional[type[BaseException]], Optional[BaseException], Optional[TracebackType]
- ]
+_ExcInfoTuple = tuple[
+ type[BaseException] | None, BaseException | None, TracebackType | None
+]
_NON_OWNED_IOLOOPS = AsyncIOMainLoop
if typing.TYPE_CHECKING:
# Additional imports only used in type comments.
# This lets us make these imports lazy.
- import datetime # noqa: F401
- from types import TracebackType # noqa: F401
- from typing import Union # noqa: F401
- import unittest # noqa: F401
+ import datetime
+ from types import TracebackType
+ import unittest
# Aliases for types that are spelled differently in different Python
# versions. bytes_type is deprecated and no longer used in Tornado
from collections.abc import Callable
from collections.abc import Awaitable, Iterable, Generator
from types import TracebackType
-import typing
-
-if typing.TYPE_CHECKING:
- from typing import Set # noqa: F401
-
# The following types are accepted by RequestHandler.set_header
# and related methods.
from tornado.util import _websocket_mask
from typing import (
- TYPE_CHECKING,
cast,
Any,
Optional,
Union,
Type,
+ Protocol,
)
from collections.abc import Callable
from collections.abc import Awaitable
from types import TracebackType
-if TYPE_CHECKING:
- from typing_extensions import Protocol
-
- # The zlib compressor types aren't actually exposed anywhere
- # publicly, so declare protocols for the portions we use.
- class _Compressor(Protocol):
- def compress(self, data: bytes) -> bytes:
- pass
-
- def flush(self, mode: int) -> bytes:
- pass
-
- class _Decompressor(Protocol):
- @property
- def unconsumed_tail(self) -> bytes:
- pass
-
- def decompress(self, data: bytes, max_length: int) -> bytes:
- pass
-
- class _WebSocketDelegate(Protocol):
- # The common base interface implemented by WebSocketHandler on
- # the server side and WebSocketClientConnection on the client
- # side.
- def on_ws_connection_close(
- self, close_code: int | None = None, close_reason: str | None = None
- ) -> None:
- pass
-
- def on_message(self, message: str | bytes) -> Optional["Awaitable[None]"]:
- pass
-
- def on_ping(self, data: bytes) -> None:
- pass
-
- def on_pong(self, data: bytes) -> None:
- pass
-
- def log_exception(
- self,
- typ: type[BaseException] | None,
- value: BaseException | None,
- tb: TracebackType | None,
- ) -> None:
- pass
+
+# The zlib compressor types aren't actually exposed anywhere
+# publicly, so declare protocols for the portions we use.
+class _Compressor(Protocol):
+ def compress(self, data: bytes) -> bytes:
+ pass
+
+ def flush(self, mode: int) -> bytes:
+ pass
+
+
+class _Decompressor(Protocol):
+ @property
+ def unconsumed_tail(self) -> bytes:
+ pass
+
+ def decompress(self, data: bytes, max_length: int) -> bytes:
+ pass
+
+
+class _WebSocketDelegate(Protocol):
+ # The common base interface implemented by WebSocketHandler on
+ # the server side and WebSocketClientConnection on the client
+ # side.
+ def on_ws_connection_close(
+ self, close_code: int | None = None, close_reason: str | None = None
+ ) -> None:
+ pass
+
+ def on_message(self, message: str | bytes) -> Optional["Awaitable[None]"]:
+ pass
+
+ def on_ping(self, data: bytes) -> None:
+ pass
+
+ def on_pong(self, data: bytes) -> None:
+ pass
+
+ def log_exception(
+ self,
+ typ: type[BaseException] | None,
+ value: BaseException | None,
+ tb: TracebackType | None,
+ ) -> None:
+ pass
_default_max_message_size = 10 * 1024 * 1024
from tornado.ioloop import IOLoop
from tornado.log import access_log
-from typing import Optional, Any
+from typing import Any
from collections.abc import Callable
from types import TracebackType
import typing
if typing.TYPE_CHECKING:
- from typing import Type # noqa: F401
- from _typeshed.wsgi import WSGIApplication as WSGIAppType # noqa: F401
+ from _typeshed.wsgi import WSGIApplication as WSGIAppType
# PEP 3333 specifies that WSGI on python 3 generally deals with byte strings
headers: list[tuple[str, str]],
exc_info: None | (
tuple[
- "Optional[Type[BaseException]]",
+ type[BaseException] | None,
BaseException | None,
TracebackType | None,
]
projects / thirdparty / tornado.git / commitdiff
