mirror of
https://github.com/Textualize/textual.git
synced 2025-10-17 02:38:12 +03:00
[Python compatibility] Make the code work with Python 3.7+
This commit is contained in:
@@ -448,7 +448,7 @@ class Compositor:
|
||||
segment_lines: list[list[Segment]] = [
|
||||
sum(
|
||||
[line for line in bucket.values() if line is not None],
|
||||
start=[],
|
||||
[],
|
||||
)
|
||||
for bucket in chops
|
||||
]
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import weakref
|
||||
from asyncio import (
|
||||
get_event_loop,
|
||||
@@ -84,7 +85,7 @@ class Timer:
|
||||
Returns:
|
||||
Task: A Task instance for the timer.
|
||||
"""
|
||||
self._task = get_event_loop().create_task(self._run())
|
||||
self._task = asyncio.create_task(self._run())
|
||||
return self._task
|
||||
|
||||
async def stop(self) -> None:
|
||||
@@ -114,7 +115,12 @@ class Timer:
|
||||
continue
|
||||
wait_time = max(0, next_timer - monotonic())
|
||||
if wait_time:
|
||||
await sleep(wait_time)
|
||||
try:
|
||||
await sleep(wait_time)
|
||||
except asyncio.CancelledError:
|
||||
# Likely our program terminating: this is fine, we just have to
|
||||
# shut down out asyncio Task properly:
|
||||
await self.stop()
|
||||
event = events.Timer(
|
||||
self.sender,
|
||||
timer=self,
|
||||
|
||||
@@ -54,6 +54,9 @@ WINDOWS = PLATFORM == "Windows"
|
||||
# asyncio will warn against resources not being cleared
|
||||
warnings.simplefilter("always", ResourceWarning)
|
||||
|
||||
# `asyncio.get_event_loop()` is deprecated since Python 3.10:
|
||||
_ASYNCIO_GET_EVENT_LOOP_IS_DEPRECATED = sys.version_info >= (3, 10, 0)
|
||||
|
||||
LayoutDefinition = "dict[str, Any]"
|
||||
|
||||
|
||||
@@ -109,6 +112,8 @@ class App(Generic[ReturnType], DOMNode):
|
||||
css (str | None, optional): CSS code to parse, or ``None`` for no literal CSS. Defaults to None.
|
||||
watch_css (bool, optional): Watch CSS for changes. Defaults to True.
|
||||
"""
|
||||
App._init_uvloop()
|
||||
|
||||
self.console = Console(
|
||||
file=sys.__stdout__, markup=False, highlight=False, emoji=False
|
||||
)
|
||||
@@ -319,28 +324,33 @@ class App(Generic[ReturnType], DOMNode):
|
||||
keys, action, description, show=show, key_display=key_display
|
||||
)
|
||||
|
||||
def run(self, loop: AbstractEventLoop | None = None) -> ReturnType | None:
|
||||
def run(self) -> ReturnType | None:
|
||||
async def run_app() -> None:
|
||||
await self.process_messages()
|
||||
|
||||
if loop:
|
||||
asyncio.set_event_loop(loop)
|
||||
else:
|
||||
try:
|
||||
import uvloop
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
asyncio.set_event_loop(uvloop.new_event_loop())
|
||||
|
||||
event_loop = asyncio.get_event_loop()
|
||||
try:
|
||||
if _ASYNCIO_GET_EVENT_LOOP_IS_DEPRECATED:
|
||||
# N.B. This doesn't work with Python<3.10, as we end up with 2 event loops:
|
||||
asyncio.run(run_app())
|
||||
finally:
|
||||
event_loop.close()
|
||||
else:
|
||||
# However, this works with Python<3.10:
|
||||
event_loop = asyncio.get_event_loop()
|
||||
event_loop.run_until_complete(run_app())
|
||||
|
||||
return self._return_value
|
||||
|
||||
@classmethod
|
||||
def _init_uvloop(cls) -> None:
|
||||
if hasattr(cls, "__uvloop_installed"):
|
||||
return
|
||||
cls.__uvloop_installed = False
|
||||
try:
|
||||
import uvloop
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
uvloop.install()
|
||||
cls.__uvloop_installed = True
|
||||
|
||||
async def _on_css_change(self) -> None:
|
||||
"""Called when the CSS changes (if watch_css is True)."""
|
||||
if self.css_file is not None:
|
||||
@@ -508,6 +518,7 @@ class App(Generic[ReturnType], DOMNode):
|
||||
active_app.set(self)
|
||||
log("---")
|
||||
log(f"driver={self.driver_class}")
|
||||
log(f"uvloop installed: {self.__class__.__uvloop_installed!r}")
|
||||
|
||||
if self.devtools_enabled:
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user