Modernize logging (#5068)

This commit is contained in:
Sebastian Rittau
2021-03-23 03:26:43 +01:00
committed by GitHub
parent 758224b231
commit 0cea325bc7
7 changed files with 793 additions and 609 deletions

View File

@@ -0,0 +1,292 @@
import threading
from _typeshed import StrPath
from string import Template
from time import struct_time
from types import FrameType, TracebackType
from typing import (
IO,
Any,
Callable,
Dict,
Iterable,
List,
Mapping,
MutableMapping,
Optional,
Sequence,
Text,
Tuple,
Union,
overload,
)
_SysExcInfoType = Union[Tuple[type, BaseException, Optional[TracebackType]], Tuple[None, None, None]]
_ExcInfoType = Union[None, bool, _SysExcInfoType]
_ArgsType = Union[Tuple[Any, ...], Mapping[str, Any]]
_FilterType = Union[Filter, Callable[[LogRecord], int]]
_Level = Union[int, Text]
raiseExceptions: bool
logThreads: bool
logMultiprocessing: bool
logProcesses: bool
_srcfile: Optional[str]
def currentframe() -> FrameType: ...
_levelNames: Dict[Union[int, str], Union[str, int]] # Union[int:str, str:int]
class Filterer(object):
filters: List[Filter]
def __init__(self) -> None: ...
def addFilter(self, filter: _FilterType) -> None: ...
def removeFilter(self, filter: _FilterType) -> None: ...
def filter(self, record: LogRecord) -> bool: ...
class Logger(Filterer):
name: str
level: int
parent: Union[Logger, PlaceHolder]
propagate: bool
handlers: List[Handler]
disabled: int
def __init__(self, name: str, level: _Level = ...) -> None: ...
def setLevel(self, level: _Level) -> None: ...
def isEnabledFor(self, level: int) -> bool: ...
def getEffectiveLevel(self) -> int: ...
def getChild(self, suffix: str) -> Logger: ...
def debug(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def info(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def warning(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
warn = warning
def error(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def critical(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
fatal = critical
def log(
self,
level: int,
msg: Any,
*args: Any,
exc_info: _ExcInfoType = ...,
extra: Optional[Dict[str, Any]] = ...,
**kwargs: Any,
) -> None: ...
def exception(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def _log(
self,
level: int,
msg: Any,
args: _ArgsType,
exc_info: Optional[_ExcInfoType] = ...,
extra: Optional[Dict[str, Any]] = ...,
) -> None: ... # undocumented
def filter(self, record: LogRecord) -> bool: ...
def addHandler(self, hdlr: Handler) -> None: ...
def removeHandler(self, hdlr: Handler) -> None: ...
def findCaller(self) -> Tuple[str, int, str]: ...
def handle(self, record: LogRecord) -> None: ...
def makeRecord(
self,
name: str,
level: int,
fn: str,
lno: int,
msg: Any,
args: _ArgsType,
exc_info: Optional[_SysExcInfoType],
func: Optional[str] = ...,
extra: Optional[Mapping[str, Any]] = ...,
) -> LogRecord: ...
CRITICAL: int
FATAL: int
ERROR: int
WARNING: int
WARN: int
INFO: int
DEBUG: int
NOTSET: int
class Handler(Filterer):
level: int # undocumented
formatter: Optional[Formatter] # undocumented
lock: Optional[threading.Lock] # undocumented
name: Optional[str] # undocumented
def __init__(self, level: _Level = ...) -> None: ...
def createLock(self) -> None: ...
def acquire(self) -> None: ...
def release(self) -> None: ...
def setLevel(self, level: _Level) -> None: ...
def setFormatter(self, fmt: Formatter) -> None: ...
def filter(self, record: LogRecord) -> bool: ...
def flush(self) -> None: ...
def close(self) -> None: ...
def handle(self, record: LogRecord) -> None: ...
def handleError(self, record: LogRecord) -> None: ...
def format(self, record: LogRecord) -> str: ...
def emit(self, record: LogRecord) -> None: ...
class Formatter:
converter: Callable[[Optional[float]], struct_time]
_fmt: Optional[str]
datefmt: Optional[str]
def __init__(self, fmt: Optional[str] = ..., datefmt: Optional[str] = ...) -> None: ...
def format(self, record: LogRecord) -> str: ...
def formatTime(self, record: LogRecord, datefmt: Optional[str] = ...) -> str: ...
def formatException(self, ei: _SysExcInfoType) -> str: ...
class Filter:
def __init__(self, name: str = ...) -> None: ...
def filter(self, record: LogRecord) -> bool: ...
class LogRecord:
args: _ArgsType
asctime: str
created: int
exc_info: Optional[_SysExcInfoType]
exc_text: Optional[str]
filename: str
funcName: str
levelname: str
levelno: int
lineno: int
module: str
msecs: int
message: str
msg: str
name: str
pathname: str
process: int
processName: str
relativeCreated: int
thread: int
threadName: str
def __init__(
self,
name: str,
level: int,
pathname: str,
lineno: int,
msg: Any,
args: _ArgsType,
exc_info: Optional[_SysExcInfoType],
func: Optional[str] = ...,
) -> None: ...
def getMessage(self) -> str: ...
class LoggerAdapter:
logger: Logger
extra: Mapping[str, Any]
def __init__(self, logger: Logger, extra: Mapping[str, Any]) -> None: ...
def process(self, msg: Any, kwargs: MutableMapping[str, Any]) -> Tuple[Any, MutableMapping[str, Any]]: ...
def debug(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def info(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def warning(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def error(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def exception(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def critical(
self, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def log(
self,
level: int,
msg: Any,
*args: Any,
exc_info: _ExcInfoType = ...,
extra: Optional[Dict[str, Any]] = ...,
**kwargs: Any,
) -> None: ...
def isEnabledFor(self, level: int) -> bool: ...
@overload
def getLogger() -> Logger: ...
@overload
def getLogger(name: Union[Text, str]) -> Logger: ...
def getLoggerClass() -> type: ...
def debug(msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any) -> None: ...
def info(msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any) -> None: ...
def warning(msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any) -> None: ...
warn = warning
def error(msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any) -> None: ...
def critical(
msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def exception(
msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
def log(
level: int, msg: Any, *args: Any, exc_info: _ExcInfoType = ..., extra: Optional[Dict[str, Any]] = ..., **kwargs: Any
) -> None: ...
fatal = critical
def disable(level: int) -> None: ...
def addLevelName(level: int, levelName: str) -> None: ...
def getLevelName(level: Union[int, str]) -> Any: ...
def makeLogRecord(dict: Mapping[str, Any]) -> LogRecord: ...
@overload
def basicConfig() -> None: ...
@overload
def basicConfig(
*,
filename: Optional[str] = ...,
filemode: str = ...,
format: str = ...,
datefmt: Optional[str] = ...,
level: Optional[_Level] = ...,
stream: IO[str] = ...,
) -> None: ...
def shutdown(handlerList: Sequence[Any] = ...) -> None: ... # handlerList is undocumented
def setLoggerClass(klass: type) -> None: ...
def captureWarnings(capture: bool) -> None: ...
class StreamHandler(Handler):
stream: IO[str] # undocumented
def __init__(self, stream: Optional[IO[str]] = ...) -> None: ...
class FileHandler(StreamHandler):
baseFilename: str # undocumented
mode: str # undocumented
encoding: Optional[str] # undocumented
delay: bool # undocumented
def __init__(self, filename: StrPath, mode: str = ..., encoding: Optional[str] = ..., delay: bool = ...) -> None: ...
def _open(self) -> IO[Any]: ...
class NullHandler(Handler): ...
class PlaceHolder:
def __init__(self, alogger: Logger) -> None: ...
def append(self, alogger: Logger) -> None: ...
# Below aren't in module docs but still visible
class RootLogger(Logger):
def __init__(self, level: int) -> None: ...
root: RootLogger
BASIC_FORMAT: str

View File

@@ -0,0 +1,14 @@
from _typeshed import AnyPath, StrPath
from threading import Thread
from typing import IO, Any, Callable, Dict, Optional, Union
from ConfigParser import RawConfigParser
_Path = StrPath
def dictConfig(config: Dict[str, Any]) -> None: ...
def fileConfig(
fname: Union[str, IO[str]], defaults: Optional[Dict[str, str]] = ..., disable_existing_loggers: bool = ...
) -> None: ...
def listen(port: int = ...) -> Thread: ...
def stopListening() -> None: ...

View File

@@ -0,0 +1,133 @@
import datetime
import ssl
from _typeshed import StrPath
from logging import FileHandler, Handler, LogRecord
from Queue import Queue
from socket import SocketKind, SocketType
from typing import Any, Callable, ClassVar, Dict, List, Optional, Tuple, Union
DEFAULT_TCP_LOGGING_PORT: int
DEFAULT_UDP_LOGGING_PORT: int
DEFAULT_HTTP_LOGGING_PORT: int
DEFAULT_SOAP_LOGGING_PORT: int
SYSLOG_UDP_PORT: int
SYSLOG_TCP_PORT: int
class WatchedFileHandler(FileHandler):
dev: int
ino: int
def __init__(self, filename: StrPath, mode: str = ..., encoding: Optional[str] = ..., delay: bool = ...) -> None: ...
def _statstream(self) -> None: ...
class RotatingFileHandler(Handler):
def __init__(
self,
filename: str,
mode: str = ...,
maxBytes: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
) -> None: ...
def doRollover(self) -> None: ...
class TimedRotatingFileHandler(Handler):
def __init__(
self,
filename: str,
when: str = ...,
interval: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
utc: bool = ...,
) -> None: ...
def doRollover(self) -> None: ...
class SocketHandler(Handler):
retryStart: float
retryFactor: float
retryMax: float
def __init__(self, host: str, port: int) -> None: ...
def makeSocket(self, timeout: float = ...) -> SocketType: ... # timeout is undocumented
def makePickle(self, record: LogRecord) -> bytes: ...
def send(self, s: bytes) -> None: ...
def createSocket(self) -> None: ...
class DatagramHandler(SocketHandler):
def makeSocket(self) -> SocketType: ... # type: ignore
class SysLogHandler(Handler):
LOG_EMERG: int
LOG_ALERT: int
LOG_CRIT: int
LOG_ERR: int
LOG_WARNING: int
LOG_NOTICE: int
LOG_INFO: int
LOG_DEBUG: int
LOG_KERN: int
LOG_USER: int
LOG_MAIL: int
LOG_DAEMON: int
LOG_AUTH: int
LOG_SYSLOG: int
LOG_LPR: int
LOG_NEWS: int
LOG_UUCP: int
LOG_CRON: int
LOG_AUTHPRIV: int
LOG_FTP: int
LOG_LOCAL0: int
LOG_LOCAL1: int
LOG_LOCAL2: int
LOG_LOCAL3: int
LOG_LOCAL4: int
LOG_LOCAL5: int
LOG_LOCAL6: int
LOG_LOCAL7: int
unixsocket: bool # undocumented
socktype: SocketKind # undocumented
facility: int # undocumented
priority_names: ClassVar[Dict[str, int]] # undocumented
facility_names: ClassVar[Dict[str, int]] # undocumented
priority_map: ClassVar[Dict[str, str]] # undocumented
def __init__(
self, address: Union[Tuple[str, int], str] = ..., facility: int = ..., socktype: Optional[SocketKind] = ...
) -> None: ...
def encodePriority(self, facility: Union[int, str], priority: Union[int, str]) -> int: ...
def mapPriority(self, levelName: str) -> str: ...
class NTEventLogHandler(Handler):
def __init__(self, appname: str, dllname: Optional[str] = ..., logtype: str = ...) -> None: ...
def getEventCategory(self, record: LogRecord) -> int: ...
# TODO correct return value?
def getEventType(self, record: LogRecord) -> int: ...
def getMessageID(self, record: LogRecord) -> int: ...
class SMTPHandler(Handler):
# TODO `secure` can also be an empty tuple
def __init__(
self,
mailhost: Union[str, Tuple[str, int]],
fromaddr: str,
toaddrs: List[str],
subject: str,
credentials: Optional[Tuple[str, str]] = ...,
secure: Union[Tuple[str], Tuple[str, str], None] = ...,
) -> None: ...
def getSubject(self, record: LogRecord) -> str: ...
class BufferingHandler(Handler):
buffer: List[LogRecord]
def __init__(self, capacity: int) -> None: ...
def shouldFlush(self, record: LogRecord) -> bool: ...
class MemoryHandler(BufferingHandler):
def __init__(self, capacity: int, flushLevel: int = ..., target: Optional[Handler] = ...) -> None: ...
def setTarget(self, target: Handler) -> None: ...
class HTTPHandler(Handler):
def __init__(self, host: str, url: str, method: str = ...) -> None: ...
def mapLogRecord(self, record: LogRecord) -> Dict[str, Any]: ...

View File

@@ -125,7 +125,7 @@ keyword: 2.7
lib2to3: 2.7
linecache: 2.7
locale: 2.7
logging: 2.7
logging: 3.6
lzma: 3.6
macpath: 2.7
macurl2path: 3.6

File diff suppressed because it is too large Load Diff

View File

@@ -1,32 +1,20 @@
import sys
from _typeshed import AnyPath, StrPath
from collections.abc import Callable
from configparser import RawConfigParser
from threading import Thread
from typing import IO, Any, Callable, Dict, Optional, Union
if sys.version_info >= (3,):
from configparser import RawConfigParser
else:
from ConfigParser import RawConfigParser
from typing import IO, Any, Optional, Union
if sys.version_info >= (3, 7):
_Path = AnyPath
else:
_Path = StrPath
def dictConfig(config: Dict[str, Any]) -> None: ...
if sys.version_info >= (3, 4):
def fileConfig(
fname: Union[_Path, IO[str], RawConfigParser],
defaults: Optional[Dict[str, str]] = ...,
disable_existing_loggers: bool = ...,
) -> None: ...
def listen(port: int = ..., verify: Optional[Callable[[bytes], Optional[bytes]]] = ...) -> Thread: ...
else:
def fileConfig(
fname: Union[str, IO[str]], defaults: Optional[Dict[str, str]] = ..., disable_existing_loggers: bool = ...
) -> None: ...
def listen(port: int = ...) -> Thread: ...
def dictConfig(config: dict[str, Any]) -> None: ...
def fileConfig(
fname: Union[_Path, IO[str], RawConfigParser],
defaults: Optional[dict[str, str]] = ...,
disable_existing_loggers: bool = ...,
) -> None: ...
def listen(port: int = ..., verify: Optional[Callable[[bytes], Optional[bytes]]] = ...) -> Thread: ...
def stopListening() -> None: ...

View File

@@ -2,16 +2,15 @@ import datetime
import ssl
import sys
from _typeshed import StrPath
from collections.abc import Callable
from logging import FileHandler, Handler, LogRecord
from socket import SocketKind, SocketType
from typing import Any, Callable, ClassVar, Dict, List, Optional, Tuple, Union
from typing import Any, ClassVar, Optional, Union
if sys.version_info >= (3, 7):
from queue import Queue, SimpleQueue
elif sys.version_info >= (3,):
from queue import Queue
else:
from Queue import Queue
from queue import Queue
DEFAULT_TCP_LOGGING_PORT: int
DEFAULT_UDP_LOGGING_PORT: int
@@ -26,90 +25,45 @@ class WatchedFileHandler(FileHandler):
def __init__(self, filename: StrPath, mode: str = ..., encoding: Optional[str] = ..., delay: bool = ...) -> None: ...
def _statstream(self) -> None: ...
if sys.version_info >= (3,):
class BaseRotatingHandler(FileHandler):
terminator: str
namer: Optional[Callable[[str], str]]
rotator: Optional[Callable[[str, str], None]]
def __init__(self, filename: StrPath, mode: str, encoding: Optional[str] = ..., delay: bool = ...) -> None: ...
def rotation_filename(self, default_name: str) -> None: ...
def rotate(self, source: str, dest: str) -> None: ...
class BaseRotatingHandler(FileHandler):
terminator: str
namer: Optional[Callable[[str], str]]
rotator: Optional[Callable[[str, str], None]]
def __init__(self, filename: StrPath, mode: str, encoding: Optional[str] = ..., delay: bool = ...) -> None: ...
def rotation_filename(self, default_name: str) -> None: ...
def rotate(self, source: str, dest: str) -> None: ...
if sys.version_info >= (3,):
class RotatingFileHandler(BaseRotatingHandler):
def __init__(
self,
filename: StrPath,
mode: str = ...,
maxBytes: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
) -> None: ...
def doRollover(self) -> None: ...
class RotatingFileHandler(BaseRotatingHandler):
def __init__(
self,
filename: StrPath,
mode: str = ...,
maxBytes: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
) -> None: ...
def doRollover(self) -> None: ...
else:
class RotatingFileHandler(Handler):
def __init__(
self,
filename: str,
mode: str = ...,
maxBytes: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
) -> None: ...
def doRollover(self) -> None: ...
if sys.version_info >= (3,):
class TimedRotatingFileHandler(BaseRotatingHandler):
if sys.version_info >= (3, 4):
def __init__(
self,
filename: StrPath,
when: str = ...,
interval: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
utc: bool = ...,
atTime: Optional[datetime.datetime] = ...,
) -> None: ...
else:
def __init__(
self,
filename: str,
when: str = ...,
interval: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
utc: bool = ...,
) -> None: ...
def doRollover(self) -> None: ...
else:
class TimedRotatingFileHandler(Handler):
def __init__(
self,
filename: str,
when: str = ...,
interval: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
utc: bool = ...,
) -> None: ...
def doRollover(self) -> None: ...
class TimedRotatingFileHandler(BaseRotatingHandler):
def __init__(
self,
filename: StrPath,
when: str = ...,
interval: int = ...,
backupCount: int = ...,
encoding: Optional[str] = ...,
delay: bool = ...,
utc: bool = ...,
atTime: Optional[datetime.datetime] = ...,
) -> None: ...
def doRollover(self) -> None: ...
class SocketHandler(Handler):
retryStart: float
retryFactor: float
retryMax: float
if sys.version_info >= (3, 4):
def __init__(self, host: str, port: Optional[int]) -> None: ...
else:
def __init__(self, host: str, port: int) -> None: ...
def __init__(self, host: str, port: Optional[int]) -> None: ...
def makeSocket(self, timeout: float = ...) -> SocketType: ... # timeout is undocumented
def makePickle(self, record: LogRecord) -> bytes: ...
def send(self, s: bytes) -> None: ...
@@ -157,14 +111,13 @@ class SysLogHandler(Handler):
LOG_LOCAL7: int
unixsocket: bool # undocumented
socktype: SocketKind # undocumented
if sys.version_info >= (3,):
ident: str # undocumented
ident: str # undocumented
facility: int # undocumented
priority_names: ClassVar[Dict[str, int]] # undocumented
facility_names: ClassVar[Dict[str, int]] # undocumented
priority_map: ClassVar[Dict[str, str]] # undocumented
priority_names: ClassVar[dict[str, int]] # undocumented
facility_names: ClassVar[dict[str, int]] # undocumented
priority_map: ClassVar[dict[str, str]] # undocumented
def __init__(
self, address: Union[Tuple[str, int], str] = ..., facility: int = ..., socktype: Optional[SocketKind] = ...
self, address: Union[tuple[str, int], str] = ..., facility: int = ..., socktype: Optional[SocketKind] = ...
) -> None: ...
def encodePriority(self, facility: Union[int, str], priority: Union[int, str]) -> int: ...
def mapPriority(self, levelName: str) -> str: ...
@@ -178,81 +131,58 @@ class NTEventLogHandler(Handler):
class SMTPHandler(Handler):
# TODO `secure` can also be an empty tuple
if sys.version_info >= (3,):
def __init__(
self,
mailhost: Union[str, Tuple[str, int]],
fromaddr: str,
toaddrs: List[str],
subject: str,
credentials: Optional[Tuple[str, str]] = ...,
secure: Union[Tuple[str], Tuple[str, str], None] = ...,
timeout: float = ...,
) -> None: ...
else:
def __init__(
self,
mailhost: Union[str, Tuple[str, int]],
fromaddr: str,
toaddrs: List[str],
subject: str,
credentials: Optional[Tuple[str, str]] = ...,
secure: Union[Tuple[str], Tuple[str, str], None] = ...,
) -> None: ...
def __init__(
self,
mailhost: Union[str, tuple[str, int]],
fromaddr: str,
toaddrs: list[str],
subject: str,
credentials: Optional[tuple[str, str]] = ...,
secure: Union[tuple[str], tuple[str, str], None] = ...,
timeout: float = ...,
) -> None: ...
def getSubject(self, record: LogRecord) -> str: ...
class BufferingHandler(Handler):
buffer: List[LogRecord]
buffer: list[LogRecord]
def __init__(self, capacity: int) -> None: ...
def shouldFlush(self, record: LogRecord) -> bool: ...
class MemoryHandler(BufferingHandler):
if sys.version_info >= (3, 6):
def __init__(
self, capacity: int, flushLevel: int = ..., target: Optional[Handler] = ..., flushOnClose: bool = ...
) -> None: ...
else:
def __init__(self, capacity: int, flushLevel: int = ..., target: Optional[Handler] = ...) -> None: ...
def __init__(
self, capacity: int, flushLevel: int = ..., target: Optional[Handler] = ..., flushOnClose: bool = ...
) -> None: ...
def setTarget(self, target: Handler) -> None: ...
class HTTPHandler(Handler):
if sys.version_info >= (3, 5):
def __init__(
self,
host: str,
url: str,
method: str = ...,
secure: bool = ...,
credentials: Optional[tuple[str, str]] = ...,
context: Optional[ssl.SSLContext] = ...,
) -> None: ...
def mapLogRecord(self, record: LogRecord) -> dict[str, Any]: ...
class QueueHandler(Handler):
if sys.version_info >= (3, 7):
def __init__(self, queue: Union[SimpleQueue[Any], Queue[Any]]) -> None: ...
else:
def __init__(self, queue: Queue[Any]) -> None: ...
def prepare(self, record: LogRecord) -> Any: ...
def enqueue(self, record: LogRecord) -> None: ...
class QueueListener:
if sys.version_info >= (3, 7):
def __init__(
self,
host: str,
url: str,
method: str = ...,
secure: bool = ...,
credentials: Optional[Tuple[str, str]] = ...,
context: Optional[ssl.SSLContext] = ...,
) -> None: ...
elif sys.version_info >= (3,):
def __init__(
self, host: str, url: str, method: str = ..., secure: bool = ..., credentials: Optional[Tuple[str, str]] = ...
self, queue: Union[SimpleQueue[Any], Queue[Any]], *handlers: Handler, respect_handler_level: bool = ...
) -> None: ...
else:
def __init__(self, host: str, url: str, method: str = ...) -> None: ...
def mapLogRecord(self, record: LogRecord) -> Dict[str, Any]: ...
if sys.version_info >= (3,):
class QueueHandler(Handler):
if sys.version_info >= (3, 7):
def __init__(self, queue: Union[SimpleQueue[Any], Queue[Any]]) -> None: ...
else:
def __init__(self, queue: Queue[Any]) -> None: ...
def prepare(self, record: LogRecord) -> Any: ...
def enqueue(self, record: LogRecord) -> None: ...
class QueueListener:
if sys.version_info >= (3, 7):
def __init__(
self, queue: Union[SimpleQueue[Any], Queue[Any]], *handlers: Handler, respect_handler_level: bool = ...
) -> None: ...
elif sys.version_info >= (3, 5):
def __init__(self, queue: Queue[Any], *handlers: Handler, respect_handler_level: bool = ...) -> None: ...
else:
def __init__(self, queue: Queue, *handlers: Handler) -> None: ...
def dequeue(self, block: bool) -> LogRecord: ...
def prepare(self, record: LogRecord) -> Any: ...
def start(self) -> None: ...
def stop(self) -> None: ...
def enqueue_sentinel(self) -> None: ...
def __init__(self, queue: Queue[Any], *handlers: Handler, respect_handler_level: bool = ...) -> None: ...
def dequeue(self, block: bool) -> LogRecord: ...
def prepare(self, record: LogRecord) -> Any: ...
def start(self) -> None: ...
def stop(self) -> None: ...
def enqueue_sentinel(self) -> None: ...