diff --git a/DEVELOPMENT_TOOLS_AND_TIPS.md b/DEVELOPMENT_TOOLS_AND_TIPS.md
new file mode 100644
index 0000000..37ac9b9
--- /dev/null
+++ b/DEVELOPMENT_TOOLS_AND_TIPS.md
@@ -0,0 +1,38 @@
+# Recommended Extensions for editing winloop
+
+If you want all the nitty gritties about vscode extensions the owner is using the [vscode-cython](https://marketplace.visualstudio.com/items?itemName=ktnrg45.vscode-cython) is strongly recommended although deprecated and no longer being maintained, when working with C++ Extensions the [Microsoft C++ Extension](https://marketplace.visualstudio.com/items?itemName=ms-vscode.cpptools) does the job. However for any open-source and privacy enthusiests another extension might be better than this one.
+
+
+## Common Problems
+
+### How do I to ensure `Python.h` is registered?
+
+If you get a message such as `cannot open source file "Python.h"` Don't be alarmed. It's because it's not registered to be visible. The way you fix this is in the `VS Code Microsoft C++ Extension` is rather easy. A Configuration that adds the __PYTHONPATH__ Environment variable you made when you setup your first installation of python should automatically do the job.
+
+Example configuration file would be `.vscode/c_cpp_properties.json`
+```json
+{
+ "configurations": [
+ {
+ "name": "Win32",
+ "includePath": [
+ "${workspaceFolder}/**",
+ "${PYTHONPATH}/**"
+ ],
+ "defines": [
+ "_DEBUG",
+ "UNICODE",
+ "_UNICODE"
+ ],
+ "windowsSdkVersion": "10.0.19041.0",
+ "compilerPath": "cl.exe",
+ "cStandard": "c17",
+ "cppStandard": "c++17",
+ "intelliSenseMode": "windows-msvc-x64"
+ }
+ ],
+ "version": 4
+}
+```
+
+After that, `Python.h` should be within reach, problem solved.
diff --git a/README.md b/README.md
index a1f1207..c022444 100644
--- a/README.md
+++ b/README.md
@@ -1,8 +1,10 @@
-
+[](https://github.com/astral-sh/ruff)
[](https://github.com/Vizonex/IDIM-Badge)
+
+
# Winloop
[](https://badge.fury.io/py/winloop)
@@ -33,208 +35,38 @@ to compile this library on your own.
python setup.py build_ext --inplace
```
-## Reporting issues
-
-If you find any bugs with this library be sure to open up an issue in the [issuetracker](https://github.com/Vizonex/Winloop/issues). Me and other contributors will be happy try to help you figure out and diagnose your problems.
-
-## Making pull requests
-We encourage anyone to make pull-requests to Winloop, containing anything from spelling mistakes to vulnerability patches. Every little bit helps keep this library maintained and alive.
-Make sure that you are able to compile the library with the steps shown above. We plan to implement a nightly workflow to verify one's pull-request in the future.
-
-
-
-```python
-try:
- import aiohttp
- import aiohttp.web
-except ImportError:
- skip_tests = True
-else:
- skip_tests = False
-
-import asyncio
-import unittest
-import weakref
-import winloop
-import sys
-
-class TestAioHTTP(unittest.TestCase):
- def __init__(self, methodName: str = "test_aiohttp_basic_1") -> None:
- super().__init__(methodName)
-
-
- def setUp(self):
- self.loop = asyncio.get_event_loop()
-
- def test_aiohttp_basic_1(self):
- PAYLOAD = 'It Works!
' * 10000
-
- async def on_request(request):
- return aiohttp.web.Response(text=PAYLOAD)
-
- asyncio.set_event_loop(self.loop)
- app = aiohttp.web.Application()
- app.router.add_get('/', on_request)
-
- runner = aiohttp.web.AppRunner(app)
- self.loop.run_until_complete(runner.setup())
- site = aiohttp.web.TCPSite(runner, '0.0.0.0', '10000')
- self.loop.run_until_complete(site.start())
- port = site._server.sockets[0].getsockname()[1]
-
- async def test():
- # Make sure we're using the correct event loop.
- self.assertIs(asyncio.get_event_loop(), self.loop)
-
- for addr in (('localhost', port),
- ('127.0.0.1', port)):
- async with aiohttp.ClientSession() as client:
- async with client.get('http://{}:{}'.format(*addr)) as r:
- self.assertEqual(r.status, 200)
- result = await r.text()
- self.assertEqual(result, PAYLOAD)
-
- self.loop.run_until_complete(test())
- self.loop.run_until_complete(runner.cleanup())
-
- def test_aiohttp_graceful_shutdown(self):
- async def websocket_handler(request):
- ws = aiohttp.web.WebSocketResponse()
- await ws.prepare(request)
- request.app['websockets'].add(ws)
- try:
- async for msg in ws:
- await ws.send_str(msg.data)
- finally:
- request.app['websockets'].discard(ws)
- return ws
-
- async def on_shutdown(app):
- for ws in set(app['websockets']):
- await ws.close(
- code=aiohttp.WSCloseCode.GOING_AWAY,
- message='Server shutdown')
-
- asyncio.set_event_loop(self.loop)
- app = aiohttp.web.Application()
- app.router.add_get('/', websocket_handler)
- app.on_shutdown.append(on_shutdown)
- app['websockets'] = weakref.WeakSet()
-
- runner = aiohttp.web.AppRunner(app)
- self.loop.run_until_complete(runner.setup())
- site = aiohttp.web.TCPSite(runner, '0.0.0.0', '10000')
- self.loop.run_until_complete(site.start())
- port = site._server.sockets[0].getsockname()[1]
-
- async def client():
- async with aiohttp.ClientSession() as client:
- async with client.ws_connect(
- 'http://127.0.0.1:{}'.format(port)) as ws:
- await ws.send_str("hello")
- async for msg in ws:
- assert msg.data == "hello"
-
- client_task = asyncio.ensure_future(client())
-
- async def stop():
- await asyncio.sleep(0.1)
- try:
- await asyncio.wait_for(runner.cleanup(), timeout=0.1)
- except Exception as e:
- print(e)
- finally:
- try:
- client_task.cancel()
- await client_task
- except asyncio.CancelledError:
- pass
-
- self.loop.run_until_complete(stop())
-
-
+## Testing with UV
+The __uv__ package manager is the preferred method for testing __winloop__ and it's author uses `ruff` and `uv` by default.
-if __name__ == "__main__":
- # print("tesing without winloop")
- # asyncio.DefaultEventLoopPolicy = asyncio.WindowsSelectorEventLoopPolicy
- # asyncio.DefaultEventLoopPolicy = asyncio.WindowsProactorEventLoopPolicy
- unittest.main()
- # Looks like winloop might be 3x faster than the Proctor Event Loop , THAT's A HUGE IMPROVEMENT!
- print("testing again but with winloop enabled")
- winloop.install()
- unittest.main()
+```
+uv venv
```
-The benchmarks for the code above are as follows
-
-## Benchmarks
-
-### TCP Connections
--------------------
-
-| Asyncio Event Loop Policy | Time (in Seconds) |
-|-----------------------------------|-----------------------|
-| WinLoopPolicy | 0.493s |
-| WindowsProactorEventLoopPolicy | 2.510s |
-| WindowsSelectorEventLoopPolicy | 2.723s |
-
-
-That's a massive increase and jump from just TCP alone! I'll be posting more benchmarks soon, as
-I modify more of the current test suites made by uvloop...
-
-
-## How to Use Winloop with Fastapi
-
-This was a cool little script I put together just to make Fastapi that much faster to handle:
-
-```python
-
-# TODO this code example is deprecated
-from fastapi import FastAPI
-from fastapi.responses import HTMLResponse
-import winloop
-import uvicorn
-import asyncio
-import datetime
-
-app = FastAPI()
-
-@app.on_event("startup")
-def make_assertion():
- # Check to make sure that we bypassed the original eventloop Policy....
- assert isinstance(asyncio.get_event_loop_policy(), winloop.WinLoopPolicy)
-
-
-@app.get("/test")
-async def test_get_request():
- return HTMLResponse("FAST API WORKS WITH WINLOOP!
")
-
+```shell
+uv pip install winloop
+```
+If your bilding and testing it from your own fork the following should work.
+```shell
+uv pip install -e .
+```
+If your planning to setup.py build_ext commands
+you will need to install setuptools and cython
+```shell
+uv pip install cython setuptools
+```
-# starllete will use asyncio.to_thread() so that this can remain asynchronous
-@app.get("/date")
-def test_dynamic_response():
- return str(datetime.datetime.now())
+## Reporting issues
-# Although tricky to pass and is not normal, it does in fact work...
-if __name__ == "__main__":
- winloop.install()
- # Winloop's eventlooppolicy will be passed to uvicorn after this point...
- loop = asyncio.get_event_loop()
- config = uvicorn.Config(app=app,port=10000,loop=loop)
- server = uvicorn.Server(config)
- asyncio.run(server.serve())
-```
+If you find any bugs with this library be sure to open up an issue in the [issuetracker](https://github.com/Vizonex/Winloop/issues). The Winloop Contributors and the Owner when available will be happy try to help you figure out and diagnose your problems.
## How To Use Winloop When Uvloop is not available
```python
-
# Here's A small Example of using winloop when uvloop is not available to us
import sys
import aiohttp
-import asyncio
async def main():
async with aiohttp.ClientSession("https://httpbin.org") as client:
@@ -250,33 +82,34 @@ if __name__ == "__main__":
run(main())
```
+Alternatively, There's another way to do this involving a smaller resource called `winuvloop` which combines both uvloop and winloop together.
- ## TODO-List
-- [ ] In Winloop 0.2.0 or before 2026 I would like to start planning to migrate to pytest so that things could be formatted a little better.
-A Migration Python script might be needed or looked into.
-
-- [ ] Update Fastapi Example to a more recent version of fastapi
-
-- [ ] Help Wanted. We're looking for other maintainers to help us.
-
-- [x] Nightly Builds And Test Suite Workflows for anyone wanting to use newer unreleased versions. (Done, it runs now)
+```python
+import aiohttp
+import winuvloop
-- [ ] Adding in the necessary hooks for pyinstaller to compile this fast library to executable code even though hooks have been known to inflate the size of the `.exe` files. This is because calling hidden-imports for all the `__init__.py` modules might annoy some developers. (Luckily I'm aware of this issue because I've been doing this myself...) [Update, This is now pending and I hope that it passes](https://github.com/pyinstaller/pyinstaller-hooks-contrib/pull/867)
+async def main():
+ async with aiohttp.ClientSession("https://httpbin.org") as client:
+ async with client.get("/ip") as resp:
+ print(await resp.json())
-- [x] write a workflow for nightly builds if necessary for verification of pull requests.
+if __name__ == "__main__":
+ winuvloop.run(main())
+```
-- [ ] Update benchmarks (They are old) can't believe I maintained this project for over a year now...
## Videos
- By me: https://www.youtube.com/watch?v=tz9RYJ6aBZ8 (I might make a tutorial on how to use and install winloop it for those who have reading problems)
-- My Presentation and Virtual Confrence: https://www.youtube.com/watch?v=Cbb6trkKWXY
+
+- My Presentation and Virtual Confrence Because I'm too busy to go to python conventions: https://www.youtube.com/watch?v=Cbb6trkKWXY
+
## Contributing
-I put my heart and soul into this library ever since it began and any help is appreciated and means a lot to me, I have other things I wish to explore so every little bit helps
+I put my heart and soul into this library ever since it began and any help is appreciated and means a lot to me, I have other things I wish to explore so every little bit helps. No matter what I go through in life, my heart is always here.
### How Can I contribute?
-- I make and branch and make edits and then I do a pull requests before I just step in and add something new.
-This way you have time to review my additions, changes or feature beforehand.
-- Forking The library.
+- Forking the library.
+- Mirroring Winloop's improvements on uvloop's side.
+- Reviewing pull requests.
- Fixing or editing workflows.
- Finding and editing spelling mistakes.
diff --git a/winloop/__init__.py b/winloop/__init__.py
index 3210b62..2f4065b 100644
--- a/winloop/__init__.py
+++ b/winloop/__init__.py
@@ -1,14 +1,17 @@
+from __future__ import annotations # 3.9 fallback
+
import asyncio as __asyncio
import typing as _typing
import sys as _sys
import warnings as _warnings
+import collections.abc as _collections_abc
from . import includes as __includes # NOQA
from .loop import Loop as __BaseLoop # NOQA
from ._version import __version__ # NOQA
-__all__: _typing.Tuple[str, ...] = ('new_event_loop', 'run')
+__all__: tuple[str, ...] = ("new_event_loop", "run")
_AbstractEventLoop = __asyncio.AbstractEventLoop
@@ -25,16 +28,16 @@ def new_event_loop() -> Loop:
if _typing.TYPE_CHECKING:
+
def run(
- main: _typing.Coroutine[_typing.Any, _typing.Any, _T],
+ main: _collections_abc.Coroutine[_typing.Any, _typing.Any, _T],
*,
- loop_factory: _typing.Optional[
- _typing.Callable[[], Loop]
- ] = new_event_loop,
- debug: _typing.Optional[bool]=None,
+ loop_factory: _collections_abc.Callable[[], Loop] | None = new_event_loop,
+ debug: bool | None = None,
) -> _T:
"""The preferred way of running a coroutine with winloop."""
else:
+
def run(main, *, loop_factory=new_event_loop, debug=None, **run_kwargs):
"""The preferred way of running a coroutine with winloop."""
@@ -44,7 +47,7 @@ async def wrapper():
# is using `winloop.run()` intentionally.
loop = __asyncio._get_running_loop()
if not isinstance(loop, Loop):
- raise TypeError('winloop.run() uses a non-winloop event loop')
+ raise TypeError("winloop.run() uses a non-winloop event loop")
return await main
vi = _sys.version_info[:2]
@@ -54,12 +57,11 @@ async def wrapper():
if __asyncio._get_running_loop() is not None:
raise RuntimeError(
- "asyncio.run() cannot be called from a running event loop")
+ "asyncio.run() cannot be called from a running event loop"
+ )
if not __asyncio.iscoroutine(main):
- raise ValueError(
- "a coroutine was expected, got {!r}".format(main)
- )
+ raise ValueError("a coroutine was expected, got {!r}".format(main))
loop = loop_factory()
try:
@@ -71,10 +73,8 @@ async def wrapper():
try:
_cancel_all_tasks(loop)
loop.run_until_complete(loop.shutdown_asyncgens())
- if hasattr(loop, 'shutdown_default_executor'):
- loop.run_until_complete(
- loop.shutdown_default_executor()
- )
+ if hasattr(loop, "shutdown_default_executor"):
+ loop.run_until_complete(loop.shutdown_default_executor())
finally:
__asyncio.set_event_loop(None)
loop.close()
@@ -82,22 +82,18 @@ async def wrapper():
elif vi == (3, 11):
if __asyncio._get_running_loop() is not None:
raise RuntimeError(
- "asyncio.run() cannot be called from a running event loop")
+ "asyncio.run() cannot be called from a running event loop"
+ )
with __asyncio.Runner(
- loop_factory=loop_factory,
- debug=debug,
- **run_kwargs
+ loop_factory=loop_factory, debug=debug, **run_kwargs
) as runner:
return runner.run(wrapper())
else:
assert vi >= (3, 12)
return __asyncio.run(
- wrapper(),
- loop_factory=loop_factory,
- debug=debug,
- **run_kwargs
+ wrapper(), loop_factory=loop_factory, debug=debug, **run_kwargs
)
@@ -111,22 +107,22 @@ def _cancel_all_tasks(loop: _AbstractEventLoop) -> None:
for task in to_cancel:
task.cancel()
- loop.run_until_complete(
- __asyncio.gather(*to_cancel, return_exceptions=True)
- )
+ loop.run_until_complete(__asyncio.gather(*to_cancel, return_exceptions=True))
for task in to_cancel:
if task.cancelled():
continue
if task.exception() is not None:
- loop.call_exception_handler({
- 'message': 'unhandled exception during asyncio.run() shutdown',
- 'exception': task.exception(),
- 'task': task,
- })
+ loop.call_exception_handler(
+ {
+ "message": "unhandled exception during asyncio.run() shutdown",
+ "exception": task.exception(),
+ "task": task,
+ }
+ )
-_deprecated_names = ('install', 'EventLoopPolicy')
+_deprecated_names = ("install", "EventLoopPolicy")
if _sys.version_info[:2] < (3, 16):
@@ -152,8 +148,8 @@ def install() -> None:
"""
if _sys.version_info[:2] >= (3, 12):
_warnings.warn(
- 'winloop.install() is deprecated in favor of winloop.run() '
- 'starting with Python 3.12.',
+ "winloop.install() is deprecated in favor of winloop.run() "
+ "starting with Python 3.12.",
DeprecationWarning,
stacklevel=1,
)
@@ -161,7 +157,7 @@ def install() -> None:
class EventLoopPolicy(
# This is to avoid a mypy error about AbstractEventLoopPolicy
- getattr(__asyncio, 'AbstractEventLoopPolicy') # type: ignore[misc]
+ getattr(__asyncio, "AbstractEventLoopPolicy") # type: ignore[misc]
):
"""Event loop policy for winloop.
@@ -183,16 +179,12 @@ def _loop_factory(self) -> Loop:
# marked as abstract in typeshed, we have to put them in so mypy
# thinks the base methods are overridden. This is the same approach
# taken for the Windows event loop policy classes in typeshed.
- def get_child_watcher(self) -> _typing.NoReturn:
- ...
+ def get_child_watcher(self) -> _typing.NoReturn: ...
- def set_child_watcher(
- self, watcher: _typing.Any
- ) -> _typing.NoReturn:
- ...
+ def set_child_watcher(self, watcher: _typing.Any) -> _typing.NoReturn: ...
class _Local(threading.local):
- _loop: _typing.Optional[_AbstractEventLoop] = None
+ _loop: _AbstractEventLoop | None = None
def __init__(self) -> None:
self._local = self._Local()
@@ -204,15 +196,13 @@ def get_event_loop(self) -> _AbstractEventLoop:
"""
if self._local._loop is None:
raise RuntimeError(
- 'There is no current event loop in thread %r.'
+ "There is no current event loop in thread %r."
% threading.current_thread().name
)
return self._local._loop
- def set_event_loop(
- self, loop: _typing.Optional[_AbstractEventLoop]
- ) -> None:
+ def set_event_loop(self, loop: _AbstractEventLoop | None) -> None:
"""Set the event loop."""
if loop is not None and not isinstance(loop, _AbstractEventLoop):
raise TypeError(
@@ -228,6 +218,6 @@ def new_event_loop(self) -> Loop:
"""
return self._loop_factory()
- globals()['install'] = install
- globals()['EventLoopPolicy'] = EventLoopPolicy
+ globals()["install"] = install
+ globals()["EventLoopPolicy"] = EventLoopPolicy
return globals()[name]
diff --git a/winloop/loop.pyi b/winloop/loop.pyi
index 9c8c462..e315535 100644
--- a/winloop/loop.pyi
+++ b/winloop/loop.pyi
@@ -1,41 +1,37 @@
+from __future__ import annotations # 3.9 fallback
+
import asyncio
import ssl
-import sys
+from collections.abc import Awaitable, Callable, Generator, Sequence
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
-from typing import (
- IO,
- Any,
- Awaitable,
- Callable,
- Dict,
- Generator,
- List,
- Optional,
- Sequence,
- Tuple,
- TypeVar,
- Union,
- overload,
-)
+from typing import IO, Any, TypeVar, overload
-_T = TypeVar('_T')
-_Context = Dict[str, Any]
+_T = TypeVar("_T")
+_Context = dict[str, Any]
_ExceptionHandler = Callable[[asyncio.AbstractEventLoop, _Context], Any]
-_SSLContext = Union[bool, None, ssl.SSLContext]
+_SSLContext = bool | None | ssl.SSLContext
_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: ...
@@ -48,52 +44,58 @@ class Loop:
def create_future(self) -> asyncio.Future[Any]: ...
def create_task(
self,
- coro: Union[Awaitable[_T], Generator[Any, None, _T]],
+ 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: None
+ | (
+ Callable[
+ [asyncio.AbstractEventLoop, Generator[Any, None, _T]],
+ asyncio.Future[_T],
+ ]
+ ),
) -> None: ...
def get_task_factory(
self,
- ) -> Optional[
- Callable[[asyncio.AbstractEventLoop, Generator[Any, None, _T]], asyncio.Future[_T]]
- ]: ...
+ ) -> (
+ None
+ | (
+ Callable[
+ [asyncio.AbstractEventLoop, Generator[Any, None, _T]],
+ asyncio.Future[_T],
+ ]
+ )
+ ): ...
@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[Union[str, bytes]],
- port: Optional[Union[str, bytes, int]],
+ host: str | bytes | None,
+ port: str | bytes | int | None,
*,
family: int = ...,
type: int = ...,
proto: int = ...,
flags: int = ...,
- ) -> List[
- Tuple[
+ ) -> list[
+ tuple[
AddressFamily,
SocketKind,
int,
str,
- Union[Tuple[str, int], Tuple[str, int, int, int]],
+ tuple[str, int] | tuple[str, int, int, int],
]
]: ...
async def getnameinfo(
self,
- sockaddr: Union[
- Tuple[str, int],
- Tuple[str, int, int],
- Tuple[str, int, int, int]
- ],
+ sockaddr: (tuple[str, int] | tuple[str, int, int] | tuple[str, int, int, int]),
flags: int = ...,
- ) -> Tuple[str, str]: ...
+ ) -> tuple[str, str]: ...
async def start_tls(
self,
transport: asyncio.BaseTransport,
@@ -101,15 +103,15 @@ 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[Union[str, Sequence[str]]] = ...,
+ host: str | Sequence[str] | None = ...,
port: int = ...,
*,
family: int = ...,
@@ -117,10 +119,10 @@ class Loop:
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
@@ -135,10 +137,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
@@ -153,10 +155,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(
@@ -171,36 +173,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: ...
@@ -209,19 +211,23 @@ class Loop:
async def sock_recv(self, sock: socket, nbytes: int) -> bytes: ...
async def sock_recv_into(self, sock: socket, buf: bytearray) -> int: ...
async def sock_sendall(self, sock: socket, data: bytes) -> None: ...
- async def sock_accept(self, sock: socket) -> Tuple[socket, _RetAddress]: ...
+ async def sock_accept(self, sock: socket) -> tuple[socket, _RetAddress]: ...
async def sock_connect(self, sock: socket, address: _Address) -> None: ...
async def sock_recvfrom(self, sock: socket, bufsize: int) -> bytes: ...
- async def sock_recvfrom_into(self, sock: socket, buf: bytearray, nbytes: int = ...) -> int: ...
- async def sock_sendto(self, sock: socket, data: bytes, address: _Address) -> None: ...
+ async def sock_recvfrom_into(
+ self, sock: socket, buf: bytearray, nbytes: int = ...
+ ) -> int: ...
+ async def sock_sendto(
+ self, sock: socket, data: bytes, address: _Address
+ ) -> None: ...
async def connect_accepted_socket(
self,
protocol_factory: Callable[[], _ProtocolT],
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
@@ -230,7 +236,7 @@ class Loop:
async def subprocess_shell(
self,
protocol_factory: Callable[[], _ProtocolT],
- cmd: Union[bytes, str],
+ cmd: bytes | str,
*,
stdin: Any = ...,
stdout: Any = ...,
@@ -259,21 +265,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
@@ -282,7 +288,7 @@ class Loop:
transport: asyncio.BaseTransport,
file: IO[bytes],
offset: int = ...,
- count: Optional[int] = ...,
+ count: int | None = ...,
*,
fallback: bool = ...,
) -> int: ...
@@ -291,7 +297,7 @@ class Loop:
sock: socket,
file: IO[bytes],
offset: int = ...,
- count: Optional[int] = ...,
+ count: int | None = ...,
*,
- fallback: bool = ...
+ fallback: bool = ...,
) -> int: ...