diff options
| author | John Snow <jsnow@redhat.com> | 2023-06-06 13:19:11 -0400 |
|---|---|---|
| committer | John Snow <jsnow@redhat.com> | 2025-09-15 14:36:01 -0400 |
| commit | f9d2e0a3bd7ba2a693a892881f91cf53fa90cc71 (patch) | |
| tree | 157b514f31a49d39b41c4e31afda8de51066e020 /python/qemu/qmp/util.py | |
| parent | 094ded5227dc4e8dde2c78a9788bf6c90771ad85 (diff) | |
| download | focaccia-qemu-f9d2e0a3bd7ba2a693a892881f91cf53fa90cc71.tar.gz focaccia-qemu-f9d2e0a3bd7ba2a693a892881f91cf53fa90cc71.zip | |
python: backport 'drop Python3.6 workarounds'
Now that the minimum version is 3.7, drop some of the 3.6-specific hacks we've been carrying. A single remaining compatibility hack concerning 3.6's lack of @asynccontextmanager is addressed in the following commit. Signed-off-by: John Snow <jsnow@redhat.com> cherry picked from commit python-qemu-qmp@3e8e34e594cfc6b707e6f67959166acde4b421b8 Signed-off-by: John Snow <jsnow@redhat.com> Reviewed-by: Daniel P. Berrangé <berrange@redhat.com>
Diffstat (limited to 'python/qemu/qmp/util.py')
| -rw-r--r-- | python/qemu/qmp/util.py | 107 |
1 files changed, 4 insertions, 103 deletions
diff --git a/python/qemu/qmp/util.py b/python/qemu/qmp/util.py index ca6225e9cd..0b3e781373 100644 --- a/python/qemu/qmp/util.py +++ b/python/qemu/qmp/util.py @@ -1,25 +1,15 @@ """ Miscellaneous Utilities -This module provides asyncio utilities and compatibility wrappers for -Python 3.6 to provide some features that otherwise become available in -Python 3.7+. - -Various logging and debugging utilities are also provided, such as -`exception_summary()` and `pretty_traceback()`, used primarily for -adding information into the logging stream. +This module provides asyncio and various logging and debugging +utilities, such as `exception_summary()` and `pretty_traceback()`, used +primarily for adding information into the logging stream. """ import asyncio import sys import traceback -from typing import ( - Any, - Coroutine, - Optional, - TypeVar, - cast, -) +from typing import TypeVar, cast T = TypeVar('T') @@ -79,95 +69,6 @@ def bottom_half(func: T) -> T: return func -# ------------------------------- -# Section: Compatibility Wrappers -# ------------------------------- - - -def create_task(coro: Coroutine[Any, Any, T], - loop: Optional[asyncio.AbstractEventLoop] = None - ) -> 'asyncio.Future[T]': - """ - Python 3.6-compatible `asyncio.create_task` wrapper. - - :param coro: The coroutine to execute in a task. - :param loop: Optionally, the loop to create the task in. - - :return: An `asyncio.Future` object. - """ - if sys.version_info >= (3, 7): - if loop is not None: - return loop.create_task(coro) - return asyncio.create_task(coro) # pylint: disable=no-member - - # Python 3.6: - return asyncio.ensure_future(coro, loop=loop) - - -def is_closing(writer: asyncio.StreamWriter) -> bool: - """ - Python 3.6-compatible `asyncio.StreamWriter.is_closing` wrapper. - - :param writer: The `asyncio.StreamWriter` object. - :return: `True` if the writer is closing, or closed. - """ - if sys.version_info >= (3, 7): - return writer.is_closing() - - # Python 3.6: - transport = writer.transport - assert isinstance(transport, asyncio.WriteTransport) - return transport.is_closing() - - -async def wait_closed(writer: asyncio.StreamWriter) -> None: - """ - Python 3.6-compatible `asyncio.StreamWriter.wait_closed` wrapper. - - :param writer: The `asyncio.StreamWriter` to wait on. - """ - if sys.version_info >= (3, 7): - await writer.wait_closed() - return - - # Python 3.6 - transport = writer.transport - assert isinstance(transport, asyncio.WriteTransport) - - while not transport.is_closing(): - await asyncio.sleep(0) - - # This is an ugly workaround, but it's the best I can come up with. - sock = transport.get_extra_info('socket') - - if sock is None: - # Our transport doesn't have a socket? ... - # Nothing we can reasonably do. - return - - while sock.fileno() != -1: - await asyncio.sleep(0) - - -def asyncio_run(coro: Coroutine[Any, Any, T], *, debug: bool = False) -> T: - """ - Python 3.6-compatible `asyncio.run` wrapper. - - :param coro: A coroutine to execute now. - :return: The return value from the coroutine. - """ - if sys.version_info >= (3, 7): - return asyncio.run(coro, debug=debug) - - # Python 3.6 - loop = asyncio.get_event_loop() - loop.set_debug(debug) - ret = loop.run_until_complete(coro) - loop.close() - - return ret - - # ---------------------------- # Section: Logging & Debugging # ---------------------------- |