# 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."""Processors and tools specific to the `Twisted <https://twisted.org/>`_networking engine.See also :doc:`structlog's Twisted support <twisted>`."""from__future__importannotationsimportjsonimportsysfromtypingimportAny,Callable,Sequence,TextIOfromtwisted.pythonimportlogfromtwisted.python.failureimportFailurefromtwisted.python.logimportILogObserver,textFromEventDictfromzope.interfaceimportimplementerfrom._baseimportBoundLoggerBasefrom._configimport_BUILTIN_DEFAULT_PROCESSORSfrom.processorsimportJSONRendererasGenericJSONRendererfrom.typingimportEventDict,WrappedLogger[docs]classBoundLogger(BoundLoggerBase):""" Twisted-specific version of `structlog.BoundLogger`. Works exactly like the generic one except that it takes advantage of knowing the logging methods in advance. Use it like:: configure( wrapper_class=structlog.twisted.BoundLogger, ) """[docs]defmsg(self,event:str|None=None,**kw:Any)->Any:""" Process event and call ``log.msg()`` with the result. """returnself._proxy_to_logger("msg",event,**kw) [docs]deferr(self,event:str|None=None,**kw:Any)->Any:""" Process event and call ``log.err()`` with the result. """returnself._proxy_to_logger("err",event,**kw) [docs]classLoggerFactory:""" Build a Twisted logger when an *instance* is called. >>> from structlog import configure >>> from structlog.twisted import LoggerFactory >>> configure(logger_factory=LoggerFactory()) """[docs]def__call__(self,*args:Any)->WrappedLogger:""" Positional arguments are silently ignored. :rvalue: A new Twisted logger. .. versionchanged:: 0.4.0 Added support for optional positional arguments. """returnlog _FAIL_TYPES=(BaseException,Failure)def_extractStuffAndWhy(eventDict:EventDict)->tuple[Any,Any,EventDict]:""" Removes all possible *_why*s and *_stuff*s, analyzes exc_info and returns a tuple of ``(_stuff, _why, eventDict)``. **Modifies** *eventDict*! """_stuff=eventDict.pop("_stuff",None)_why=eventDict.pop("_why",None)event=eventDict.pop("event",None)ifisinstance(_stuff,_FAIL_TYPES)andisinstance(event,_FAIL_TYPES):raiseValueError("Both _stuff and event contain an Exception/Failure.")# `log.err('event', _why='alsoEvent')` is ambiguous.if_whyandisinstance(event,str):raiseValueError("Both `_why` and `event` supplied.")# Two failures are ambiguous too.ifnotisinstance(_stuff,_FAIL_TYPES)andisinstance(event,_FAIL_TYPES):_why=_whyor"error"_stuff=eventifisinstance(event,str):_why=eventifnot_stuffandsys.exc_info()!=(None,None,None):_stuff=Failure()# type: ignore[no-untyped-call]# Either we used the error ourselves or the user supplied one for# formatting. Avoid log.err() to dump another traceback into the log.ifisinstance(_stuff,BaseException)andnotisinstance(_stuff,Failure):_stuff=Failure(_stuff)# type: ignore[no-untyped-call]return_stuff,_why,eventDictclassReprWrapper:""" Wrap a string and return it as the ``__repr__``. This is needed for ``twisted.python.log.err`` that calls `repr` on ``_stuff``: >>> repr("foo") "'foo'" >>> repr(ReprWrapper("foo")) 'foo' Note the extra quotes in the unwrapped example. """def__init__(self,string:str)->None:self.string=stringdef__eq__(self,other:object)->bool:""" Check for equality, just for tests. """return(isinstance(other,self.__class__)andself.string==other.string)def__repr__(self)->str:returnself.string[docs]classJSONRenderer(GenericJSONRenderer):""" Behaves like `structlog.processors.JSONRenderer` except that it formats tracebacks and failures itself if called with ``err()``. .. note:: This ultimately means that the messages get logged out using ``msg()``, and *not* ``err()`` which renders failures in separate lines. Therefore it will break your tests that contain assertions using `flushLoggedErrors <https://docs.twisted.org/en/stable/api/ twisted.trial.unittest.SynchronousTestCase.html#flushLoggedErrors>`_. *Not* an adapter like `EventAdapter` but a real formatter. Also does *not* require to be adapted using it. Use together with a `JSONLogObserverWrapper`-wrapped Twisted logger like `plainJSONStdOutLogger` for pure-JSON logs. """def__call__(# type: ignore[override]self,logger:WrappedLogger,name:str,eventDict:EventDict,)->tuple[Sequence[Any],dict[str,Any]]:_stuff,_why,eventDict=_extractStuffAndWhy(eventDict)ifname=="err":eventDict["event"]=_whyifisinstance(_stuff,Failure):eventDict["exception"]=_stuff.getTraceback(detail="verbose")_stuff.cleanFailure()# type: ignore[no-untyped-call]else:eventDict["event"]=_whyreturn((ReprWrapper(GenericJSONRenderer.__call__(# type: ignore[arg-type]self,logger,name,eventDict)),),{"_structlog":True},) [docs]@implementer(ILogObserver)classPlainFileLogObserver:""" Write only the plain message without timestamps or anything else. Great to just print JSON to stdout where you catch it with something like runit. Args: file: File to print to. .. versionadded:: 0.2.0 """def__init__(self,file:TextIO)->None:self._write=file.writeself._flush=file.flushdef__call__(self,eventDict:EventDict)->None:self._write(textFromEventDict(eventDict)# type: ignore[arg-type, operator]+"\n",)self._flush() [docs]@implementer(ILogObserver)classJSONLogObserverWrapper:""" Wrap a log *observer* and render non-`JSONRenderer` entries to JSON. Args: observer (ILogObserver): Twisted log observer to wrap. For example :class:`PlainFileObserver` or Twisted's stock `FileLogObserver <https://docs.twisted.org/en/stable/api/ twisted.python.log.FileLogObserver.html>`_ .. versionadded:: 0.2.0 """def__init__(self,observer:Any)->None:self._observer=observerdef__call__(self,eventDict:EventDict)->str:if"_structlog"notineventDict:eventDict["message"]=(json.dumps({"event":textFromEventDict(eventDict# type: ignore[arg-type]),"system":eventDict.get("system"),}),)eventDict["_structlog"]=Truereturnself._observer(eventDict) [docs]defplainJSONStdOutLogger()->JSONLogObserverWrapper:""" Return a logger that writes only the message to stdout. Transforms non-`JSONRenderer` messages to JSON. Ideal for JSONifying log entries from Twisted plugins and libraries that are outside of your control:: $ twistd -n --logger structlog.twisted.plainJSONStdOutLogger web {"event": "Log opened.", "system": "-"} {"event": "twistd 13.1.0 (python 2.7.3) starting up.", "system": "-"} {"event": "reactor class: twisted...EPollReactor.", "system": "-"} {"event": "Site starting on 8080", "system": "-"} {"event": "Starting factory <twisted.web.server.Site ...>", ...} ... Composes `PlainFileLogObserver` and `JSONLogObserverWrapper` to a usable logger. .. versionadded:: 0.2.0 """returnJSONLogObserverWrapper(PlainFileLogObserver(sys.stdout)) [docs]classEventAdapter:""" Adapt an ``event_dict`` to Twisted logging system. Particularly, make a wrapped `twisted.python.log.err <https://docs.twisted.org/en/stable/api/twisted.python.log.html#err>`_ behave as expected. Args: dictRenderer: Renderer that is used for the actual log message. Please note that structlog comes with a dedicated `JSONRenderer`. **Must** be the last processor in the chain and requires a *dictRenderer* for the actual formatting as an constructor argument in order to be able to fully support the original behaviors of ``log.msg()`` and ``log.err()``. """def__init__(self,dictRenderer:(Callable[[WrappedLogger,str,EventDict],str]|None)=None,)->None:self._dictRenderer=dictRendereror_BUILTIN_DEFAULT_PROCESSORS[-1]def__call__(self,logger:WrappedLogger,name:str,eventDict:EventDict)->Any:ifname=="err":# This aspires to handle the following cases correctly:# 1. log.err(failure, _why='event', **kw)# 2. log.err('event', **kw)# 3. log.err(_stuff=failure, _why='event', **kw)_stuff,_why,eventDict=_extractStuffAndWhy(eventDict)eventDict["event"]=_whyreturn((),{"_stuff":_stuff,"_why":self._dictRenderer(logger,name,eventDict),},)returnself._dictRenderer(logger,name,eventDict)
