|
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677 |
- # -*- test-case-name: twisted.logger.test.test_file -*-
- # Copyright (c) Twisted Matrix Laboratories.
- # See LICENSE for details.
-
- """
- File log observer.
- """
-
- from typing import IO, Any, Callable, Optional
-
- from zope.interface import implementer
-
- from twisted.python.compat import ioType
- from ._format import formatEventAsClassicLogText, formatTime, timeFormatRFC3339
- from ._interfaces import ILogObserver, LogEvent
-
-
- @implementer(ILogObserver)
- class FileLogObserver:
- """
- Log observer that writes to a file-like object.
- """
-
- def __init__(
- self, outFile: IO[Any], formatEvent: Callable[[LogEvent], Optional[str]]
- ) -> None:
- """
- @param outFile: A file-like object. Ideally one should be passed which
- accepts text data. Otherwise, UTF-8 L{bytes} will be used.
- @param formatEvent: A callable that formats an event.
- """
- if ioType(outFile) is not str:
- self._encoding: Optional[str] = "utf-8"
- else:
- self._encoding = None
-
- self._outFile = outFile
- self.formatEvent = formatEvent
-
- def __call__(self, event: LogEvent) -> None:
- """
- Write event to file.
-
- @param event: An event.
- """
- text = self.formatEvent(event)
-
- if text:
- if self._encoding is None:
- self._outFile.write(text)
- else:
- self._outFile.write(text.encode(self._encoding))
- self._outFile.flush()
-
-
- def textFileLogObserver(
- outFile: IO[Any], timeFormat: Optional[str] = timeFormatRFC3339
- ) -> FileLogObserver:
- """
- Create a L{FileLogObserver} that emits text to a specified (writable)
- file-like object.
-
- @param outFile: A file-like object. Ideally one should be passed which
- accepts text data. Otherwise, UTF-8 L{bytes} will be used.
- @param timeFormat: The format to use when adding timestamp prefixes to
- logged events. If L{None}, or for events with no C{"log_timestamp"}
- key, the default timestamp prefix of C{"-"} is used.
-
- @return: A file log observer.
- """
-
- def formatEvent(event: LogEvent) -> Optional[str]:
- return formatEventAsClassicLogText(
- event, formatTime=lambda e: formatTime(e, timeFormat)
- )
-
- return FileLogObserver(outFile, formatEvent)
|