# SPDX-License-Identifier: MIT OR Apache-2.0# This file is dual licensed under the terms of the Apache License, Version# 2.0, and the MIT License. See the LICENSE file in the root of this# repository for complete details."""Logger classes responsible for output."""from__future__importannotationsimportcopyimportsysimportthreadingfrompickleimportPicklingErrorfromsysimportstderr,stdoutfromtypingimportIO,Any,BinaryIO,TextIOWRITE_LOCKS:dict[IO[Any],threading.Lock]={}def_get_lock_for_file(file:IO[Any])->threading.Lock:lock=WRITE_LOCKS.get(file)iflockisNone:lock=threading.Lock()WRITE_LOCKS[file]=lockreturnlock
[docs]classPrintLogger:""" Print events into a file. Args: file: File to print to. (default: `sys.stdout`) >>> from structlog import PrintLogger >>> PrintLogger().info("hello") hello Useful if you follow `current logging best practices <logging-best-practices>`. Also very useful for testing and examples since `logging` is finicky in doctests. .. versionchanged:: 22.1.0 The implementation has been switched to use `print` for better monkeypatchability. """def__init__(self,file:TextIO|None=None):self._file=fileorstdoutself._lock=_get_lock_for_file(self._file)def__getstate__(self)->str:""" Our __getattr__ magic makes this necessary. """ifself._fileisstdout:return"stdout"ifself._fileisstderr:return"stderr"raisePicklingError("Only PrintLoggers to sys.stdout and sys.stderr can be pickled.")def__setstate__(self,state:Any)->None:""" Our __getattr__ magic makes this necessary. """ifstate=="stdout":self._file=stdoutelse:self._file=stderrself._lock=_get_lock_for_file(self._file)def__deepcopy__(self,memodict:dict[str,object])->PrintLogger:""" Create a new PrintLogger with the same attributes. Similar to pickling. """ifself._filenotin(stdout,stderr):raisecopy.error("Only PrintLoggers to sys.stdout and sys.stderr ""can be deepcopied.")newself=self.__class__(self._file)newself._lock=_get_lock_for_file(newself._file)returnnewselfdef__repr__(self)->str:returnf"<PrintLogger(file={self._file!r})>"
[docs]classPrintLoggerFactory:r""" Produce `PrintLogger`\ s. To be used with `structlog.configure`\ 's ``logger_factory``. Args: file: File to print to. (default: `sys.stdout`) Positional arguments are silently ignored. .. versionadded:: 0.4.0 """def__init__(self,file:TextIO|None=None):self._file=filedef__call__(self,*args:Any)->PrintLogger:returnPrintLogger(self._file)
[docs]classWriteLogger:""" Write events into a file. Args: file: File to print to. (default: `sys.stdout`) >>> from structlog import WriteLogger >>> WriteLogger().info("hello") hello Useful if you follow `current logging best practices <logging-best-practices>`. Also very useful for testing and examples since `logging` is finicky in doctests. A little faster and a little less versatile than `structlog.PrintLogger`. .. versionadded:: 22.1.0 """def__init__(self,file:TextIO|None=None):self._file=fileorsys.stdoutself._write=self._file.writeself._flush=self._file.flushself._lock=_get_lock_for_file(self._file)def__getstate__(self)->str:""" Our __getattr__ magic makes this necessary. """ifself._fileisstdout:return"stdout"ifself._fileisstderr:return"stderr"raisePicklingError("Only WriteLoggers to sys.stdout and sys.stderr can be pickled.")def__setstate__(self,state:Any)->None:""" Our __getattr__ magic makes this necessary. """ifstate=="stdout":self._file=stdoutelse:self._file=stderrself._lock=_get_lock_for_file(self._file)def__deepcopy__(self,memodict:dict[str,object])->WriteLogger:""" Create a new WriteLogger with the same attributes. Similar to pickling. """ifself._filenotin(sys.stdout,sys.stderr):raisecopy.error("Only WriteLoggers to sys.stdout and sys.stderr ""can be deepcopied.")newself=self.__class__(self._file)newself._write=newself._file.writenewself._flush=newself._file.flushnewself._lock=_get_lock_for_file(newself._file)returnnewselfdef__repr__(self)->str:returnf"<WriteLogger(file={self._file!r})>"
[docs]defmsg(self,message:str)->None:""" Write and flush *message*. """withself._lock:self._write(message+"\n")self._flush()
[docs]classWriteLoggerFactory:r""" Produce `WriteLogger`\ s. To be used with `structlog.configure`\ 's ``logger_factory``. Args: file: File to print to. (default: `sys.stdout`) Positional arguments are silently ignored. .. versionadded:: 22.1.0 """def__init__(self,file:TextIO|None=None):self._file=filedef__call__(self,*args:Any)->WriteLogger:returnWriteLogger(self._file)
[docs]classBytesLogger:r""" Writes bytes into a file. Args: file: File to print to. (default: `sys.stdout`\ ``.buffer``) Useful if you follow `current logging best practices <logging-best-practices>` together with a formatter that returns bytes (e.g. `orjson <https://github.com/ijl/orjson>`_). .. versionadded:: 20.2.0 """__slots__=("_file","_flush","_lock","_write")def__init__(self,file:BinaryIO|None=None):self._file=fileorsys.stdout.bufferself._write=self._file.writeself._flush=self._file.flushself._lock=_get_lock_for_file(self._file)def__getstate__(self)->str:""" Our __getattr__ magic makes this necessary. """ifself._fileissys.stdout.buffer:return"stdout"ifself._fileissys.stderr.buffer:return"stderr"raisePicklingError("Only BytesLoggers to sys.stdout and sys.stderr can be pickled.")def__setstate__(self,state:Any)->None:""" Our __getattr__ magic makes this necessary. """ifstate=="stdout":self._file=sys.stdout.bufferelse:self._file=sys.stderr.bufferself._write=self._file.writeself._flush=self._file.flushself._lock=_get_lock_for_file(self._file)def__deepcopy__(self,memodict:dict[str,object])->BytesLogger:""" Create a new BytesLogger with the same attributes. Similar to pickling. """ifself._filenotin(sys.stdout.buffer,sys.stderr.buffer):raisecopy.error("Only BytesLoggers to sys.stdout and sys.stderr ""can be deepcopied.")newself=self.__class__(self._file)newself._write=newself._file.writenewself._flush=newself._file.flushnewself._lock=_get_lock_for_file(newself._file)returnnewselfdef__repr__(self)->str:returnf"<BytesLogger(file={self._file!r})>"
[docs]classBytesLoggerFactory:r""" Produce `BytesLogger`\ s. To be used with `structlog.configure`\ 's ``logger_factory``. Args: file: File to print to. (default: `sys.stdout`\ ``.buffer``) Positional arguments are silently ignored. .. versionadded:: 20.2.0 """__slots__=("_file",)def__init__(self,file:BinaryIO|None=None):self._file=filedef__call__(self,*args:Any)->BytesLogger:returnBytesLogger(self._file)