def _init_cloud_handler()

in pathology/shared_libs/logging_lib/cloud_logging_client_instance.py [0:0]

• 58 lines of code
• 6 McCabe index (conditional complexity)

  def _init_cloud_handler(self) -> logging.Logger:
    """Initializes cloud logging handler and returns python logger."""
    # Instantiates a cloud logging client to generate text logs for cloud
    # operations
    with CloudLoggingClientInstance._global_lock:
      if not self._enabled or self._use_absl_logging:
        return logging.getLogger()  # Default PY logger
      handler_instance_init_params = (
          self._get_cloud_logging_handler_init_params()
      )
      if CloudLoggingClientInstance._cloud_logging_handler is not None:
        running_handler_init_params = (
            CloudLoggingClientInstance._cloud_logging_handler_init_params
        )
        if running_handler_init_params != handler_instance_init_params:
          # Call fork_shutdown to shutdown the process's named logging handler.
          raise CloudLoggerInstanceExceptionError(
              'Cloud logging handler is running with parameters that do not'
              ' match instance defined parameters. Running handler parameters:'
              f' {running_handler_init_params}; Instance parameters:'
              f' {handler_instance_init_params}'
          )
        return logging.getLogger(self._get_python_logger_name())
      log_name = self.log_name
      struct_log = {}
      struct_log['log_name'] = log_name
      struct_log['log_all_python_logs'] = self._log_all_python_logs_to_cloud
      try:
        # Attach default python & absl logger to also write to named log.
        logging_client = cloud_logging.Client(
            project=self._gcp_project_name if self._gcp_project_name else None,
            credentials=self._gcp_credentials,
        )
        logging_client.project = (
            self._gcp_project_name if self._gcp_project_name else None
        )
        handler = cloud_logging.handlers.CloudLoggingHandler(
            client=logging_client,
            name=log_name,
        )
        CloudLoggingClientInstance._cloud_logging_handler = handler
        CloudLoggingClientInstance._cloud_logging_handler_init_params = (
            handler_instance_init_params
        )
        cloud_logging.handlers.setup_logging(
            handler,
            log_level=logging.DEBUG
            if self._log_all_python_logs_to_cloud
            else logging.INFO,
        )
        dpas_python_logger = logging.getLogger(self._get_python_logger_name())
        dpas_python_logger.setLevel(
            logging.DEBUG
        )  # pytype: disable=attribute-error
        return dpas_python_logger
      except google.auth.exceptions.DefaultCredentialsError as exp:
        self._use_absl_logging = True
        self.error('Error initializing logging.', struct_log, exp)
        return logging.getLogger()
      except Exception as exp:
        self._use_absl_logging = True
        self.error('Error unexpected exception.', struct_log, exp)
        raise