"""Thread File Handler Class"""
# standard library
import os
import threading
from logging.handlers import RotatingFileHandler
from typing import Optional
[docs]class ThreadFileHandler(RotatingFileHandler):
"""Logger handler for ThreatConnect Exchange File logging."""
handler_key = None
thread_key = None
def __init__(
self,
filename: str,
mode: Optional[str] = 'a',
maxBytes: Optional[int] = 0,
backupCount: Optional[int] = 0,
encoding: Optional[str] = None,
delay: Optional[bool] = False,
):
"""Add logic to create log directory if it does not exists.
Args:
filename: The name of the logfile.
mode: The write mode for the file.
maxBytes: The max file size before rotating.
backupCount: The maximum # of backup files.
encoding: The log file encoding.
delay: If True, then file opening is deferred until the first call to emit().
"""
if encoding is None and os.getenv('LANG') is None:
encoding = 'UTF-8'
if not os.path.exists(os.path.dirname(filename)):
os.makedirs(os.path.dirname(filename), exist_ok=True)
RotatingFileHandler.__init__(self, filename, mode, maxBytes, backupCount, encoding, delay)
[docs] def emit(self, record: object):
"""Emit a record.
Emit logging events only if handler_key matches thread_key.
Args:
record: The record to be logged.
"""
# handler_key and thread_key are added in logger.add_thread_file_handler() method
if hasattr(threading.current_thread(), self.thread_key):
if self.handler_key == getattr(threading.current_thread(), self.thread_key):
RotatingFileHandler.emit(self, record)