Apply ruff/pyupgrade rule UP045

UP045 Use `X | None` for type annotations

Author: DimitriPapadopoulos

uvloop/loop.pyi

@@ -7,7 +7,6 @@ from typing import (
     Awaitable,
     Callable,
     Generator,
-    Optional,
     Sequence,
     TypeVar,
     overload,
@@ -21,16 +20,16 @@ _ProtocolT = TypeVar("_ProtocolT", bound=asyncio.BaseProtocol)
 
 class Loop:
     def call_soon(
-        self, callback: Callable[..., Any], *args: Any, context: Optional[Any] = ...
+        self, callback: Callable[..., Any], *args: Any, context: Any | None = ...
     ) -> asyncio.Handle: ...
     def call_soon_threadsafe(
-        self, callback: Callable[..., Any], *args: Any, context: Optional[Any] = ...
+        self, callback: Callable[..., Any], *args: Any, context: Any | None = ...
     ) -> asyncio.Handle: ...
     def call_later(
-        self, delay: float, callback: Callable[..., Any], *args: Any, context: Optional[Any] = ...
+        self, delay: float, callback: Callable[..., Any], *args: Any, context: Any | None = ...
     ) -> asyncio.TimerHandle: ...
     def call_at(
-        self, when: float, callback: Callable[..., Any], *args: Any, context: Optional[Any] = ...
+        self, when: float, callback: Callable[..., Any], *args: Any, context: Any | None = ...
     ) -> asyncio.TimerHandle: ...
     def time(self) -> float: ...
     def stop(self) -> None: ...
@@ -45,27 +44,23 @@ class Loop:
         self,
         coro: Awaitable[_T] | Generator[Any, None, _T],
         *,
-        name: Optional[str] = ...,
+        name: str | None = ...,
     ) -> asyncio.Task[_T]: ...
     def set_task_factory(
         self,
-        factory: Optional[
-            Callable[[asyncio.AbstractEventLoop, Generator[Any, None, _T]], asyncio.Future[_T]]
-        ],
+        factory: Callable[[asyncio.AbstractEventLoop, Generator[Any, None, _T]], asyncio.Future[_T]] | None,
     ) -> None: ...
     def get_task_factory(
         self,
-    ) -> Optional[
-        Callable[[asyncio.AbstractEventLoop, Generator[Any, None, _T]], asyncio.Future[_T]]
-    ]: ...
+    ) -> Callable[[asyncio.AbstractEventLoop, Generator[Any, None, _T]], asyncio.Future[_T]] | None: ...
     @overload
     def run_until_complete(self, future: Generator[Any, None, _T]) -> _T: ...
     @overload
     def run_until_complete(self, future: Awaitable[_T]) -> _T: ...
     async def getaddrinfo(
         self,
-        host: Optional[str | bytes],
-        port: Optional[str | bytes | int],
+        host: str | bytes | None,
+        port: str | bytes | int | None,
         *,
         family: int = ...,
         type: int = ...,
@@ -92,26 +87,26 @@ class Loop:
         sslcontext: ssl.SSLContext,
         *,
         server_side: bool = ...,
-        server_hostname: Optional[str] = ...,
-        ssl_handshake_timeout: Optional[float] = ...,
-        ssl_shutdown_timeout: Optional[float] = ...,
+        server_hostname: str | None = ...,
+        ssl_handshake_timeout: float | None = ...,
+        ssl_shutdown_timeout: float | None = ...,
     ) -> asyncio.BaseTransport: ...
     @overload
     async def create_server(
         self,
         protocol_factory: asyncio.events._ProtocolFactory,
-        host: Optional[str | Sequence[str]] = ...,
+        host: str | Sequence[str] | None = ...,
         port: int = ...,
         *,
         family: int = ...,
         flags: int = ...,
         sock: None = ...,
         backlog: int = ...,
         ssl: _SSLContext = ...,
-        reuse_address: Optional[bool] = ...,
-        reuse_port: Optional[bool] = ...,
-        ssl_handshake_timeout: Optional[float] = ...,
-        ssl_shutdown_timeout: Optional[float] = ...,
+        reuse_address: bool | None = ...,
+        reuse_port: bool | None = ...,
+        ssl_handshake_timeout: float | None = ...,
+        ssl_shutdown_timeout: float | None = ...,
         start_serving: bool = ...,
     ) -> asyncio.AbstractServer: ...
     @overload
@@ -126,10 +121,10 @@ class Loop:
         sock: socket = ...,
         backlog: int = ...,
         ssl: _SSLContext = ...,
-        reuse_address: Optional[bool] = ...,
-        reuse_port: Optional[bool] = ...,
-        ssl_handshake_timeout: Optional[float] = ...,
-        ssl_shutdown_timeout: Optional[float] = ...,
+        reuse_address: bool | None = ...,
+        reuse_port: bool | None = ...,
+        ssl_handshake_timeout: float | None = ...,
+        ssl_shutdown_timeout: float | None = ...,
         start_serving: bool = ...,
     ) -> asyncio.AbstractServer: ...
     @overload
@@ -144,10 +139,10 @@ class Loop:
         proto: int = ...,
         flags: int = ...,
         sock: None = ...,
-        local_addr: Optional[tuple[str, int]] = ...,
-        server_hostname: Optional[str] = ...,
-        ssl_handshake_timeout: Optional[float] = ...,
-        ssl_shutdown_timeout: Optional[float] = ...,
+        local_addr: tuple[str, int] | None = ...,
+        server_hostname: str | None = ...,
+        ssl_handshake_timeout: float | None = ...,
+        ssl_shutdown_timeout: float | None = ...,
     ) -> tuple[asyncio.BaseProtocol, _ProtocolT]: ...
     @overload
     async def create_connection(
@@ -162,36 +157,36 @@ class Loop:
         flags: int = ...,
         sock: socket,
         local_addr: None = ...,
-        server_hostname: Optional[str] = ...,
-        ssl_handshake_timeout: Optional[float] = ...,
-        ssl_shutdown_timeout: Optional[float] = ...,
+        server_hostname: str | None = ...,
+        ssl_handshake_timeout: float | None = ...,
+        ssl_shutdown_timeout: float | None = ...,
     ) -> tuple[asyncio.BaseProtocol, _ProtocolT]: ...
     async def create_unix_server(
         self,
         protocol_factory: asyncio.events._ProtocolFactory,
-        path: Optional[str] = ...,
+        path: str | None = ...,
         *,
         backlog: int = ...,
-        sock: Optional[socket] = ...,
+        sock: socket | None = ...,
         ssl: _SSLContext = ...,
-        ssl_handshake_timeout: Optional[float] = ...,
-        ssl_shutdown_timeout: Optional[float] = ...,
+        ssl_handshake_timeout: float | None = ...,
+        ssl_shutdown_timeout: float | None = ...,
         start_serving: bool = ...,
     ) -> asyncio.AbstractServer: ...
     async def create_unix_connection(
         self,
         protocol_factory: Callable[[], _ProtocolT],
-        path: Optional[str] = ...,
+        path: str | None = ...,
         *,
         ssl: _SSLContext = ...,
-        sock: Optional[socket] = ...,
-        server_hostname: Optional[str] = ...,
-        ssl_handshake_timeout: Optional[float] = ...,
-        ssl_shutdown_timeout: Optional[float] = ...,
+        sock: socket | None = ...,
+        server_hostname: str | None = ...,
+        ssl_handshake_timeout: float | None = ...,
+        ssl_shutdown_timeout: float | None = ...,
     ) -> tuple[asyncio.BaseProtocol, _ProtocolT]: ...
     def default_exception_handler(self, context: _Context) -> None: ...
-    def get_exception_handler(self) -> Optional[_ExceptionHandler]: ...
-    def set_exception_handler(self, handler: Optional[_ExceptionHandler]) -> None: ...
+    def get_exception_handler(self) -> _ExceptionHandler | None: ...
+    def set_exception_handler(self, handler: _ExceptionHandler | None) -> None: ...
     def call_exception_handler(self, context: _Context) -> None: ...
     def add_reader(self, fd: Any, callback: Callable[..., Any], *args: Any) -> None: ...
     def remove_reader(self, fd: Any) -> None: ...
@@ -211,8 +206,8 @@ class Loop:
         sock: socket,
         *,
         ssl: _SSLContext = ...,
-        ssl_handshake_timeout: Optional[float] = ...,
-        ssl_shutdown_timeout: Optional[float] = ...,
+        ssl_handshake_timeout: float | None = ...,
+        ssl_shutdown_timeout: float | None = ...,
     ) -> tuple[asyncio.BaseProtocol, _ProtocolT]: ...
     async def run_in_executor(
         self, executor: Any, func: Callable[..., _T], *args: Any
@@ -250,21 +245,21 @@ class Loop:
     async def create_datagram_endpoint(
         self,
         protocol_factory: Callable[[], _ProtocolT],
-        local_addr: Optional[tuple[str, int]] = ...,
-        remote_addr: Optional[tuple[str, int]] = ...,
+        local_addr: tuple[str, int] | None = ...,
+        remote_addr: tuple[str, int] | None = ...,
         *,
         family: int = ...,
         proto: int = ...,
         flags: int = ...,
-        reuse_address: Optional[bool] = ...,
-        reuse_port: Optional[bool] = ...,
-        allow_broadcast: Optional[bool] = ...,
-        sock: Optional[socket] = ...,
+        reuse_address: bool | None = ...,
+        reuse_port: bool | None = ...,
+        allow_broadcast: bool | None = ...,
+        sock: socket | None = ...,
     ) -> tuple[asyncio.BaseProtocol, _ProtocolT]: ...
     async def shutdown_asyncgens(self) -> None: ...
     async def shutdown_default_executor(
         self,
-        timeout: Optional[float] = ...,
+        timeout: float | None = ...,
     ) -> None: ...
     # Loop doesn't implement these, but since they are marked as abstract in typeshed,
     # we have to put them in so mypy thinks the base methods are overridden
@@ -273,7 +268,7 @@ class Loop:
         transport: asyncio.BaseTransport,
         file: IO[bytes],
         offset: int = ...,
-        count: Optional[int] = ...,
+        count: int | None = ...,
         *,
         fallback: bool = ...,
     ) -> int: ...
@@ -282,7 +277,7 @@ class Loop:
         sock: socket,
         file: IO[bytes],
         offset: int = ...,
-        count: Optional[int] = ...,
+        count: int | None = ...,
         *,
         fallback: bool = ...
     ) -> int: ...