From 9ba9bf3f93df4587e15281b635cbf48577216822 Mon Sep 17 00:00:00 2001 From: Delgan Date: Sat, 7 Dec 2024 17:55:24 +0100 Subject: [PATCH] wip --- loguru/__init__.py | 4 + loguru/__init__.pyi | 368 -------------------------------------------- loguru/_logger.py | 111 +++++++++++-- loguru/_types.py | 61 ++++++++ pyproject.toml | 3 +- 5 files changed, 164 insertions(+), 383 deletions(-) delete mode 100644 loguru/__init__.pyi create mode 100644 loguru/_types.py diff --git a/loguru/__init__.py b/loguru/__init__.py index 1bcd8244..9f71d18f 100644 --- a/loguru/__init__.py +++ b/loguru/__init__.py @@ -6,11 +6,15 @@ import atexit as _atexit import sys as _sys +import typing as _typing from . import _defaults from ._logger import Core as _Core from ._logger import Logger as _Logger +if _typing.TYPE_CHECKING: + from ._logger import Logger as Logger + __version__ = "0.7.3" __all__ = ["logger"] diff --git a/loguru/__init__.pyi b/loguru/__init__.pyi deleted file mode 100644 index 3a13ad71..00000000 --- a/loguru/__init__.pyi +++ /dev/null @@ -1,368 +0,0 @@ -import sys -from asyncio import AbstractEventLoop -from datetime import datetime, time, timedelta -from logging import Handler -from multiprocessing.context import BaseContext -from types import TracebackType -from typing import ( - Any, - BinaryIO, - Callable, - Dict, - Generator, - Generic, - List, - NamedTuple, - NewType, - Optional, - Pattern, - Sequence, - TextIO, - Tuple, - Type, - TypeVar, - Union, - overload, -) - -if sys.version_info >= (3, 6): - from typing import Awaitable -else: - from typing_extensions import Awaitable - -if sys.version_info >= (3, 6): - from os import PathLike - from typing import ContextManager - - PathLikeStr = PathLike[str] -else: - from pathlib import PurePath as PathLikeStr - - from typing_extensions import ContextManager - -if sys.version_info >= (3, 8): - from typing import Protocol, TypedDict -else: - from typing_extensions import Protocol, TypedDict - -_T = TypeVar("_T") -_F = TypeVar("_F", bound=Callable[..., Any]) -ExcInfo = Tuple[Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]] - -class _GeneratorContextManager(ContextManager[_T], Generic[_T]): - def __call__(self, func: _F) -> _F: ... - def __exit__( - self, - typ: Optional[Type[BaseException]], - value: Optional[BaseException], - traceback: Optional[TracebackType], - ) -> Optional[bool]: ... - -Catcher = NewType("Catcher", _GeneratorContextManager[None]) -Contextualizer = NewType("Contextualizer", _GeneratorContextManager[None]) -AwaitableCompleter = Awaitable[None] - -class Level(NamedTuple): - name: str - no: int - color: str - icon: str - -class _RecordAttribute: - def __format__(self, spec: str) -> str: ... - -class RecordFile(_RecordAttribute): - name: str - path: str - -class RecordLevel(_RecordAttribute): - name: str - no: int - icon: str - -class RecordThread(_RecordAttribute): - id: int - name: str - -class RecordProcess(_RecordAttribute): - id: int - name: str - -class RecordException(NamedTuple): - type: Optional[Type[BaseException]] - value: Optional[BaseException] - traceback: Optional[TracebackType] - -class Record(TypedDict): - elapsed: timedelta - exception: Optional[RecordException] - extra: Dict[Any, Any] - file: RecordFile - function: str - level: RecordLevel - line: int - message: str - module: str - name: Optional[str] - process: RecordProcess - thread: RecordThread - time: datetime - -class Message(str): - record: Record - -class Writable(Protocol): - def write(self, message: Message) -> None: ... - -FilterDict = Dict[Optional[str], Union[str, int, bool]] -FilterFunction = Callable[[Record], bool] -FormatFunction = Callable[[Record], str] -PatcherFunction = Callable[[Record], None] -RotationFunction = Callable[[Message, TextIO], bool] -RetentionFunction = Callable[[List[str]], None] -CompressionFunction = Callable[[str], None] - -StandardOpener = Callable[[str, int], int] - -class BasicHandlerConfig(TypedDict, total=False): - sink: Union[TextIO, Writable, Callable[[Message], None], Handler] - level: Union[str, int] - format: Union[str, FormatFunction] - filter: Optional[Union[str, FilterFunction, FilterDict]] - colorize: Optional[bool] - serialize: bool - backtrace: bool - diagnose: bool - enqueue: bool - catch: bool - -class FileHandlerConfig(TypedDict, total=False): - sink: Union[str, PathLikeStr] - level: Union[str, int] - format: Union[str, FormatFunction] - filter: Optional[Union[str, FilterFunction, FilterDict]] - colorize: Optional[bool] - serialize: bool - backtrace: bool - diagnose: bool - enqueue: bool - catch: bool - rotation: Optional[Union[str, int, time, timedelta, RotationFunction]] - retention: Optional[Union[str, int, timedelta, RetentionFunction]] - compression: Optional[Union[str, CompressionFunction]] - delay: bool - watch: bool - mode: str - buffering: int - encoding: str - errors: Optional[str] - newline: Optional[str] - closefd: bool - opener: Optional[StandardOpener] - -class AsyncHandlerConfig(TypedDict, total=False): - sink: Callable[[Message], Awaitable[None]] - level: Union[str, int] - format: Union[str, FormatFunction] - filter: Optional[Union[str, FilterFunction, FilterDict]] - colorize: Optional[bool] - serialize: bool - backtrace: bool - diagnose: bool - enqueue: bool - catch: bool - context: Optional[Union[str, BaseContext]] - loop: Optional[AbstractEventLoop] - -HandlerConfig = Union[BasicHandlerConfig, FileHandlerConfig, AsyncHandlerConfig] - -class LevelConfig(TypedDict, total=False): - name: str - no: int - color: str - icon: str - -ActivationConfig = Tuple[Optional[str], bool] - -class Logger: - @overload - def add( - self, - sink: Union[TextIO, Writable, Callable[[Message], None], Handler], - *, - level: Union[str, int] = ..., - format: Union[str, FormatFunction] = ..., - filter: Optional[Union[str, FilterFunction, FilterDict]] = ..., - colorize: Optional[bool] = ..., - serialize: bool = ..., - backtrace: bool = ..., - diagnose: bool = ..., - enqueue: bool = ..., - context: Optional[Union[str, BaseContext]] = ..., - catch: bool = ... - ) -> int: ... - @overload - def add( - self, - sink: Callable[[Message], Awaitable[None]], - *, - level: Union[str, int] = ..., - format: Union[str, FormatFunction] = ..., - filter: Optional[Union[str, FilterFunction, FilterDict]] = ..., - colorize: Optional[bool] = ..., - serialize: bool = ..., - backtrace: bool = ..., - diagnose: bool = ..., - enqueue: bool = ..., - catch: bool = ..., - context: Optional[Union[str, BaseContext]] = ..., - loop: Optional[AbstractEventLoop] = ... - ) -> int: ... - @overload - def add( - self, - sink: Union[str, PathLikeStr], - *, - level: Union[str, int] = ..., - format: Union[str, FormatFunction] = ..., - filter: Optional[Union[str, FilterFunction, FilterDict]] = ..., - colorize: Optional[bool] = ..., - serialize: bool = ..., - backtrace: bool = ..., - diagnose: bool = ..., - enqueue: bool = ..., - context: Optional[Union[str, BaseContext]] = ..., - catch: bool = ..., - rotation: Optional[Union[str, int, time, timedelta, RotationFunction]] = ..., - retention: Optional[Union[str, int, timedelta, RetentionFunction]] = ..., - compression: Optional[Union[str, CompressionFunction]] = ..., - delay: bool = ..., - watch: bool = ..., - mode: str = ..., - buffering: int = ..., - encoding: str = ..., - errors: Optional[str] = ..., - newline: Optional[str] = ..., - closefd: bool = ..., - opener: Optional[StandardOpener] = ..., - ) -> int: ... - def remove(self, handler_id: Optional[int] = ...) -> None: ... - def complete(self) -> AwaitableCompleter: ... - @overload - def catch( - self, - exception: Union[Type[BaseException], Tuple[Type[BaseException], ...]] = ..., - *, - level: Union[str, int] = ..., - reraise: bool = ..., - onerror: Optional[Callable[[BaseException], None]] = ..., - exclude: Optional[Union[Type[BaseException], Tuple[Type[BaseException], ...]]] = ..., - default: Any = ..., - message: str = ... - ) -> Catcher: ... - @overload - def catch(self, function: _F) -> _F: ... - def opt( - self, - *, - exception: Optional[Union[bool, ExcInfo, BaseException]] = ..., - record: bool = ..., - lazy: bool = ..., - colors: bool = ..., - raw: bool = ..., - capture: bool = ..., - depth: int = ..., - ansi: bool = ... - ) -> Logger: ... - def bind(__self, **kwargs: Any) -> Logger: ... # noqa: N805 - def contextualize(__self, **kwargs: Any) -> Contextualizer: ... # noqa: N805 - def patch(self, patcher: PatcherFunction) -> Logger: ... - @overload - def level(self, name: str) -> Level: ... - @overload - def level( - self, name: str, no: int = ..., color: Optional[str] = ..., icon: Optional[str] = ... - ) -> Level: ... - @overload - def level( - self, - name: str, - no: Optional[int] = ..., - color: Optional[str] = ..., - icon: Optional[str] = ..., - ) -> Level: ... - def disable(self, name: Optional[str]) -> None: ... - def enable(self, name: Optional[str]) -> None: ... - def configure( - self, - *, - handlers: Optional[Sequence[HandlerConfig]] = ..., - levels: Optional[Sequence[LevelConfig]] = ..., - extra: Optional[Dict[Any, Any]] = ..., - patcher: Optional[PatcherFunction] = ..., - activation: Optional[Sequence[ActivationConfig]] = ... - ) -> List[int]: ... - # @staticmethod cannot be used with @overload in mypy (python/mypy#7781). - # However Logger is not exposed and logger is an instance of Logger - # so for type checkers it is all the same whether it is defined here - # as a static method or an instance method. - @overload - def parse( - self, - file: Union[str, PathLikeStr, TextIO], - pattern: Union[str, Pattern[str]], - *, - cast: Union[Dict[str, Callable[[str], Any]], Callable[[Dict[str, str]], None]] = ..., - chunk: int = ... - ) -> Generator[Dict[str, Any], None, None]: ... - @overload - def parse( - self, - file: BinaryIO, - pattern: Union[bytes, Pattern[bytes]], - *, - cast: Union[Dict[str, Callable[[bytes], Any]], Callable[[Dict[str, bytes]], None]] = ..., - chunk: int = ... - ) -> Generator[Dict[str, Any], None, None]: ... - @overload - def trace(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 - @overload - def trace(__self, __message: Any) -> None: ... # noqa: N805 - @overload - def debug(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 - @overload - def debug(__self, __message: Any) -> None: ... # noqa: N805 - @overload - def info(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 - @overload - def info(__self, __message: Any) -> None: ... # noqa: N805 - @overload - def success(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 - @overload - def success(__self, __message: Any) -> None: ... # noqa: N805 - @overload - def warning(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 - @overload - def warning(__self, __message: Any) -> None: ... # noqa: N805 - @overload - def error(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 - @overload - def error(__self, __message: Any) -> None: ... # noqa: N805 - @overload - def critical(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 - @overload - def critical(__self, __message: Any) -> None: ... # noqa: N805 - @overload - def exception(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 - @overload - def exception(__self, __message: Any) -> None: ... # noqa: N805 - @overload - def log( - __self, __level: Union[int, str], __message: str, *args: Any, **kwargs: Any # noqa: N805 - ) -> None: ... - @overload - def log(__self, __level: Union[int, str], __message: Any) -> None: ... # noqa: N805 - def start(self, *args: Any, **kwargs: Any) -> int: ... - def stop(self, *args: Any, **kwargs: Any) -> None: ... - -logger: Logger diff --git a/loguru/_logger.py b/loguru/_logger.py index 0167a1eb..c9421eb0 100644 --- a/loguru/_logger.py +++ b/loguru/_logger.py @@ -98,6 +98,7 @@ from multiprocessing.context import BaseContext from os.path import basename, splitext from threading import current_thread +from typing import Any, Dict, List, Optional, Tuple, Union, overload from . import _asyncio_loop, _colorama, _defaults, _filters from ._better_exceptions import ExceptionFormatter @@ -111,6 +112,7 @@ from ._locks_machinery import create_logger_lock from ._recattrs import RecordException, RecordFile, RecordLevel, RecordProcess, RecordThread from ._simple_sinks import AsyncSink, CallableSink, StandardSink, StreamSink +from ._types import ExcInfo, PatcherFunction if sys.version_info >= (3, 6): from os import PathLike @@ -120,6 +122,7 @@ Level = namedtuple("Level", ["name", "no", "color", "icon"]) # noqa: PYI024 + start_time = aware_now() context = ContextVar("loguru_context", default={}) @@ -210,6 +213,19 @@ def __setstate__(self, state): self.lock = create_logger_lock() +Options = Tuple[ + Optional[Union[bool, ExcInfo, BaseException]], + int, + bool, + bool, + bool, + bool, + bool, + List[PatcherFunction], + Dict[Any, Any], +] + + class Logger: """An object to dispatch logging messages to configured handlers. @@ -233,7 +249,19 @@ class Logger: You should not instantiate a |Logger| by yourself, use ``from loguru import logger`` instead. """ - def __init__(self, core, exception, depth, record, lazy, colors, raw, capture, patchers, extra): + def __init__( + self, + core: Core, + exception: Optional[Union[bool, ExcInfo, BaseException]], + depth: int, + record: bool, + lazy: bool, + colors: bool, + raw: bool, + capture: bool, + patchers: List[PatcherFunction], + extra: Dict[Any, Any], + ): self._core = core self._options = (exception, depth, record, lazy, colors, raw, capture, patchers, extra) @@ -1017,7 +1045,7 @@ def add( return handler_id - def remove(self, handler_id=None): + def remove(self, handler_id: Optional[int] = None) -> None: """Remove a previously added handler and stop sending logs to its sink. Parameters @@ -1928,7 +1956,15 @@ def _find_iter(fileobj, regex, chunk): buffer = buffer[end:] yield from matches[:-1] - def _log(self, level, from_decorator, options, message, args, kwargs): + def _log( + self, + level: Union[str, int], + from_decorator: bool, + options: Options, + message: Any, + args: Tuple[Any], + kwargs: Dict[str, Any], + ): core = self._core if not core.handlers: @@ -2065,44 +2101,91 @@ def _log(self, level, from_decorator, options, message, args, kwargs): for handler in core.handlers.values(): handler.emit(log_record, level_id, from_decorator, raw, colored_message) - def trace(__self, __message, *args, **kwargs): # noqa: N805 + @overload + def trace(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 + @overload + def trace(__self, __message: Any) -> None: ... # noqa: N805 + + def trace(__self, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log ``message.format(*args, **kwargs)`` with severity ``'TRACE'``.""" __self._log("TRACE", False, __self._options, __message, args, kwargs) - def debug(__self, __message, *args, **kwargs): # noqa: N805 + @overload + def debug(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 + @overload + def debug(__self, __message: Any) -> None: ... # noqa: N805 + + def debug(__self, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log ``message.format(*args, **kwargs)`` with severity ``'DEBUG'``.""" __self._log("DEBUG", False, __self._options, __message, args, kwargs) - def info(__self, __message, *args, **kwargs): # noqa: N805 + @overload + def info(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 + @overload + def info(__self, __message: Any) -> None: ... # noqa: N805 + + def info(__self, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log ``message.format(*args, **kwargs)`` with severity ``'INFO'``.""" __self._log("INFO", False, __self._options, __message, args, kwargs) - def success(__self, __message, *args, **kwargs): # noqa: N805 + @overload + def success(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 + @overload + def success(__self, __message: Any) -> None: ... # noqa: N805 + + def success(__self, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log ``message.format(*args, **kwargs)`` with severity ``'SUCCESS'``.""" __self._log("SUCCESS", False, __self._options, __message, args, kwargs) - def warning(__self, __message, *args, **kwargs): # noqa: N805 + @overload + def warning(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 + @overload + def warning(__self, __message: Any) -> None: ... # noqa: N805 + + def warning(__self, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log ``message.format(*args, **kwargs)`` with severity ``'WARNING'``.""" __self._log("WARNING", False, __self._options, __message, args, kwargs) - def error(__self, __message, *args, **kwargs): # noqa: N805 + @overload + def error(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 + @overload + def error(__self, __message: Any) -> None: ... # noqa: N805 + + def error(__self, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log ``message.format(*args, **kwargs)`` with severity ``'ERROR'``.""" __self._log("ERROR", False, __self._options, __message, args, kwargs) - def critical(__self, __message, *args, **kwargs): # noqa: N805 + @overload + def critical(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 + @overload + def critical(__self, __message: Any) -> None: ... # noqa: N805 + + def critical(__self, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log ``message.format(*args, **kwargs)`` with severity ``'CRITICAL'``.""" __self._log("CRITICAL", False, __self._options, __message, args, kwargs) - def exception(__self, __message, *args, **kwargs): # noqa: N805 + @overload + def exception(__self, __message: str, *args: Any, **kwargs: Any) -> None: ... # noqa: N805 + @overload + def exception(__self, __message: Any) -> None: ... # noqa: N805 + + def exception(__self, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log an ``'ERROR'```` message while also capturing the currently handled exception.""" options = (True,) + __self._options[1:] __self._log("ERROR", False, options, __message, args, kwargs) - def log(__self, __level, __message, *args, **kwargs): # noqa: N805 + @overload + def log( + __self, __level: Union[int, str], __message: str, *args: Any, **kwargs: Any # noqa: N805 + ) -> None: ... + @overload + def log(__self, __level: Union[int, str], __message: Any) -> None: ... # noqa: N805 + + def log(__self, __level: str | int, __message: Any, *args: Any, **kwargs: Any): # noqa: N805 r"""Log ``message.format(*args, **kwargs)`` with severity ``level``.""" __self._log(__level, False, __self._options, __message, args, kwargs) - def start(self, *args, **kwargs): + def start(self, *args: Any, **kwargs: Any) -> int: """Add a handler sending log messages to a sink adequately configured. Deprecated function, use |add| instead. @@ -2120,7 +2203,7 @@ def start(self, *args, **kwargs): ) return self.add(*args, **kwargs) - def stop(self, *args, **kwargs): + def stop(self, *args: Any, **kwargs: Any) -> None: """Remove a previously added handler and stop sending logs to its sink. Deprecated function, use |remove| instead. diff --git a/loguru/_types.py b/loguru/_types.py new file mode 100644 index 00000000..f44d99c8 --- /dev/null +++ b/loguru/_types.py @@ -0,0 +1,61 @@ +import sys +from datetime import datetime, timedelta +from types import TracebackType +from typing import Any, Callable, Dict, NamedTuple, Optional, Tuple, Type + +if sys.version_info >= (3, 8): + from typing import TypedDict +else: + from typing_extensions import TypedDict + + +class _RecordAttribute: + def __format__(self, spec: str) -> str: ... + + +class RecordFile(_RecordAttribute): + name: str + path: str + + +class RecordLevel(_RecordAttribute): + name: str + no: int + icon: str + + +class RecordThread(_RecordAttribute): + id: int + name: str + + +class RecordProcess(_RecordAttribute): + id: int + name: str + + +class RecordException(NamedTuple): + type: Optional[Type[BaseException]] + value: Optional[BaseException] + traceback: Optional[TracebackType] + + +class Record(TypedDict): + elapsed: timedelta + exception: Optional[RecordException] + extra: Dict[Any, Any] + file: RecordFile + function: str + level: RecordLevel + line: int + message: str + module: str + name: Optional[str] + process: RecordProcess + thread: RecordThread + time: datetime + + +ExcInfo = Tuple[Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]] + +PatcherFunction = Callable[[Record], None] diff --git a/pyproject.toml b/pyproject.toml index 4ea6eb8e..5d66fc54 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,7 @@ dependencies = [ "colorama>=0.3.4 ; sys_platform=='win32'", "aiocontextvars>=0.2.0 ; python_version<'3.7'", "win32-setctime>=1.0.0 ; sys_platform=='win32'", + "typing_extensions>=3.7.4 ; python_version<'3.8'", ] description = "Python logging made (stupidly) simple" dynamic = ['version'] @@ -87,7 +88,7 @@ target-version = ["py35"] # Types are defined in a stub file. Unfortunately, type checkers such as Pyright and Mypy are # unable to "merge" them with the file containing the actual Python implementation. This causes # many false positives, therefore type checking is disabled to avoid noisy errors in the editor. -typeCheckingMode = "off" +typeCheckingMode = "strict" [tool.flit.module] name = "loguru"