classify-split-extract-workflow/classify-job/logging_handler.py (31 lines of code) (raw):

# Copyright 2024 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # pylint: disable=import-error,too-few-public-methods """ Helper Logger """ import logging import os import sys import google.cloud.logging CLOUD_LOGGING_ENABLED = True if CLOUD_LOGGING_ENABLED: client = google.cloud.logging.Client() client.setup_logging() logging.basicConfig( format="%(asctime)s:%(levelname)s:%(message)s", level=logging.INFO ) else: logging.basicConfig(level=logging.INFO) class Logger: """Class for handling logs.""" def __init__(self, name): """ Initialize the Logger instance. Args: name (str): The name of the logger, typically the module name. """ dirname = os.path.dirname(name) filename = os.path.split(name)[1] folder = os.path.split(dirname)[1] module_name = f"{folder}/{filename}" self.logger = logging.getLogger(module_name) handler = logging.StreamHandler(sys.stdout) log_format = ( "%(levelname)s: [%(name)s:%(lineno)d - " "%(funcName)s()] %(message)s" ) handler.setFormatter(logging.Formatter(log_format)) self.logger.addHandler(handler) self.logger.propagate = False @classmethod def get_logger(cls, name) -> logging.Logger: """ Get a logger instance. Args: name (str): The name of the logger, typically the module name. Returns: logging.Logger: A configured logger instance. """ logger_instance = cls(name) return logger_instance.logger