mirror of
https://github.com/Textualize/textual.git
synced 2025-10-17 02:38:12 +03:00
@@ -1,7 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import inspect
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import Callable, TYPE_CHECKING
|
||||
|
||||
import rich.repr
|
||||
from rich.console import RenderableType
|
||||
@@ -9,8 +10,24 @@ from rich.console import RenderableType
|
||||
__all__ = ["log", "panic"]
|
||||
|
||||
|
||||
from ._context import active_app
|
||||
from ._log import LogGroup, LogVerbosity, LogSeverity
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .app import App
|
||||
|
||||
if sys.version_info >= (3, 10):
|
||||
from typing import TypeAlias
|
||||
else: # pragma: no cover
|
||||
from typing_extensions import TypeAlias
|
||||
|
||||
|
||||
LogCallable: TypeAlias = "Callable"
|
||||
|
||||
|
||||
class LoggerError(Exception):
|
||||
"""Raised when the logger failed."""
|
||||
|
||||
|
||||
@rich.repr.auto
|
||||
class Logger:
|
||||
@@ -18,32 +35,46 @@ class Logger:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
log_callable: LogCallable | None,
|
||||
group: LogGroup = LogGroup.INFO,
|
||||
verbosity: LogVerbosity = LogVerbosity.NORMAL,
|
||||
severity: LogSeverity = LogSeverity.NORMAL,
|
||||
) -> None:
|
||||
self._log = log_callable
|
||||
self._group = group
|
||||
self._verbosity = verbosity
|
||||
self._severity = severity
|
||||
|
||||
@property
|
||||
def log(self) -> LogCallable:
|
||||
if self._log is None:
|
||||
try:
|
||||
app = active_app.get()
|
||||
except LookupError:
|
||||
raise LoggerError("Unable to log without an active app.") from None
|
||||
return app._log
|
||||
return self._log
|
||||
|
||||
def __rich_repr__(self) -> rich.repr.Result:
|
||||
yield self._group, LogGroup.INFO
|
||||
yield self._verbosity, LogVerbosity.NORMAL
|
||||
yield self._severity, LogSeverity.NORMAL
|
||||
|
||||
def __call__(self, *args: object, **kwargs) -> None:
|
||||
from ._context import active_app
|
||||
|
||||
app = active_app.get()
|
||||
caller = inspect.stack()[1]
|
||||
app._log(
|
||||
self._group,
|
||||
self._verbosity,
|
||||
self._severity,
|
||||
*args,
|
||||
_textual_calling_frame=caller,
|
||||
**kwargs,
|
||||
)
|
||||
try:
|
||||
self.log(
|
||||
self._group,
|
||||
self._verbosity,
|
||||
self._severity,
|
||||
*args,
|
||||
_textual_calling_frame=caller,
|
||||
**kwargs,
|
||||
)
|
||||
except LoggerError:
|
||||
# If there is not active app, try printing
|
||||
print_args = (*args, *[f"{key}={value!r}" for key, value in kwargs.items()])
|
||||
print(*print_args)
|
||||
|
||||
def verbosity(self, verbose: bool) -> Logger:
|
||||
"""Get a new logger with selective verbosity.
|
||||
@@ -55,50 +86,50 @@ class Logger:
|
||||
Logger: New logger.
|
||||
"""
|
||||
verbosity = LogVerbosity.HIGH if verbose else LogVerbosity.NORMAL
|
||||
return Logger(self._group, verbosity, LogSeverity.NORMAL)
|
||||
return Logger(self._log, self._group, verbosity, LogSeverity.NORMAL)
|
||||
|
||||
@property
|
||||
def verbose(self) -> Logger:
|
||||
"""A verbose logger."""
|
||||
return Logger(self._group, LogVerbosity.HIGH)
|
||||
return Logger(self._log, self._group, LogVerbosity.HIGH)
|
||||
|
||||
@property
|
||||
def critical(self) -> Logger:
|
||||
"""A critical logger."""
|
||||
return Logger(self._group, self._verbosity, LogSeverity.CRITICAL)
|
||||
return Logger(self._log, self._group, self._verbosity, LogSeverity.CRITICAL)
|
||||
|
||||
@property
|
||||
def event(self) -> Logger:
|
||||
"""An event logger."""
|
||||
return Logger(LogGroup.EVENT)
|
||||
return Logger(self._log, LogGroup.EVENT)
|
||||
|
||||
@property
|
||||
def debug(self) -> Logger:
|
||||
"""A debug logger."""
|
||||
return Logger(LogGroup.DEBUG)
|
||||
return Logger(self._log, LogGroup.DEBUG)
|
||||
|
||||
@property
|
||||
def info(self) -> Logger:
|
||||
"""An info logger."""
|
||||
return Logger(LogGroup.INFO)
|
||||
return Logger(self._log, LogGroup.INFO)
|
||||
|
||||
@property
|
||||
def warning(self) -> Logger:
|
||||
"""An info logger."""
|
||||
return Logger(LogGroup.WARNING)
|
||||
return Logger(self._log, LogGroup.WARNING)
|
||||
|
||||
@property
|
||||
def error(self) -> Logger:
|
||||
"""An error logger."""
|
||||
return Logger(LogGroup.ERROR)
|
||||
return Logger(self._log, LogGroup.ERROR)
|
||||
|
||||
@property
|
||||
def system(self) -> Logger:
|
||||
"""A system logger."""
|
||||
return Logger(LogGroup.SYSTEM)
|
||||
return Logger(self._log, LogGroup.SYSTEM)
|
||||
|
||||
|
||||
log = Logger()
|
||||
log = Logger(None)
|
||||
|
||||
|
||||
def panic(*args: RenderableType) -> None:
|
||||
|
||||
@@ -197,7 +197,7 @@ class App(Generic[ReturnType], DOMNode):
|
||||
else:
|
||||
self._title = title
|
||||
|
||||
self._logger = Logger()
|
||||
self._logger = Logger(self._log)
|
||||
|
||||
self._bindings.bind("ctrl+c", "quit", show=False, allow_forward=False)
|
||||
self._refresh_required = False
|
||||
|
||||
@@ -220,11 +220,9 @@ class EventMonitor(threading.Thread):
|
||||
self.target = target
|
||||
self.exit_event = exit_event
|
||||
self.process_event = process_event
|
||||
self.app.log("event monitor constructed")
|
||||
super().__init__()
|
||||
|
||||
def run(self) -> None:
|
||||
self.app.log("event monitor thread started")
|
||||
exit_requested = self.exit_event.is_set
|
||||
parser = XTermParser(self.target, lambda: False)
|
||||
|
||||
@@ -280,8 +278,7 @@ class EventMonitor(threading.Thread):
|
||||
self.on_size_change(*new_size)
|
||||
|
||||
except Exception as error:
|
||||
self.app.log("EVENT MONITOR ERROR", error)
|
||||
self.app.log("event monitor thread finished")
|
||||
self.app.log.error("EVENT MONITOR ERROR", error)
|
||||
|
||||
def on_size_change(self, width: int, height: int) -> None:
|
||||
"""Called when terminal size changes."""
|
||||
|
||||
Reference in New Issue
Block a user