airbyte-cdk/python/airbyte_cdk/logger.py
def format(self, record: logging.LogRecord) -> str:
        """Return a JSON representation of the log message"""
        message = super().format(record)
        airbyte_level = self.level_mapping.get(record.levelno, "INFO")
        for secret in AirbyteLogFormatter._secrets:
            message = message.replace(secret, "****")
        log_message = AirbyteMessage(type="LOG", log=AirbyteLogMessage(level=airbyte_level, message=message))
        return log_message.json(exclude_unset=True)
Return a JSON representation of the log message
Return a JSON representation of the log message
Return a JSON representation of the log message
def format(self, record: logging.LogRecord) -> str:
        message = super().format(record)
        airbyte_level = self.level_mapping.get(record.levelno, "INFO")
        for secret in AirbyteLogFormatter._secrets:
            message = message.replace(secret, "****")
        log_message = AirbyteMessage(type="LOG", log=AirbyteLogMessage(level=airbyte_level, message=message))
        return log_message.json(exclude_unset=True)
def format ( self , record : logging . LogRecord ) -> str : message = super ( ) . format ( record ) airbyte_level = self . level_mapping . get ( record . levelno , "INFO" ) for secret in AirbyteLogFormatter . _secrets : message = message . replace ( secret , "****" ) log_message = AirbyteMessage ( type = "LOG" , log = AirbyteLogMessage ( level = airbyte_level , message = message ) ) return log_message . json ( exclude_unset = True )
Return a JSON representation of the log message
Return a JSON representation of the log message
"""Return a JSON representation of the log message"""
self
record
logging.LogRecord