feat: add automatic logging config to support debug logging

google/api_core/client_logging.py

@@ -0,0 +1,76 @@
+import logging
+import json
+import re
+import os
+
+_LOGGING_INITIALIZED = False
+_BASE_LOGGER_NAME = "google"
+# TODO(https://github.com/googleapis/python-api-core/issues/760): Update Request / Response messages.
+REQUEST_MESSAGE = "Sending request ..."
+RESPONSE_MESSAGE = "Receiving response ..."
+
+# TODO(https://github.com/googleapis/python-api-core/issues/761): Update this list to support additional logging fields
+_recognized_logging_fields = ["httpRequest", "rpcName", "serviceName"] # Additional fields to be Logged.
+
+def logger_configured(logger):
+  return logger.handlers != [] or logger.level != logging.NOTSET or logger.propagate == False
+
+def initialize_logging():
+   global _LOGGING_INITIALIZED
+   if _LOGGING_INITIALIZED:
+     return
+   scopes = os.getenv("GOOGLE_SDK_PYTHON_LOGGING_SCOPE")
+   setup_logging(scopes)
+   _LOGGING_INITIALIZED = True
+
+def parse_logging_scopes(scopes):
+  if not scopes:
+     return []
+  # TODO(https://github.com/googleapis/python-api-core/issues/759): check if the namespace is a valid namespace.
+  # TODO(b/380481951): Support logging multiple scopes.
+  # TODO(b/380483756): Raise or log a warning for an invalid scope.
+  namespaces = [scopes]
+  return namespaces
+
+def configure_defaults(logger):
+   if not logger_configured(logger):
+        console_handler = logging.StreamHandler()
+        logger.setLevel("DEBUG")
+        logger.propagate = False
+        formatter = StructuredLogFormatter()
+        console_handler.setFormatter(formatter)
+        logger.addHandler(console_handler)
+
+def setup_logging(scopes=[]):
+
+    # only returns valid logger scopes (namespaces)
+    # this list has at most one element.
+    logger_names = parse_logging_scopes(scopes) 
+
+    for namespace in logger_names:
+      # This will either create a module level logger or get the reference of the base logger instantiated above.
+      logger = logging.getLogger(namespace)
+
+      # Configure default settings.
+      configure_defaults(logger)
+
+    # disable log propagation at base logger level to the root logger only if a base logger is not already configured via code changes.
+    # Maybe we do this at the end?
+    base_logger = logging.getLogger(_BASE_LOGGER_NAME)
+    if not logger_configured(base_logger):
+        base_logger.propagate = False
+
+class StructuredLogFormatter(logging.Formatter):
+    def format(self, record):
+        log_obj = {
+            'timestamp': self.formatTime(record),
+            'severity': record.levelname,
+            'name': record.name,
+            'message': record.getMessage(),
+        }
+
+        for field_name in _recognized_logging_fields:
+            value = getattr(record, field_name, None)
+            if value is not None:
+                log_obj[field_name] = value
+        return json.dumps(log_obj)

tests/unit/test_client_logging.py

@@ -0,0 +1,62 @@
+import logging
+import pytest
+
+from google.api_core.client_logging import setup_logging
+# Test expected behaviour for warnings, propagation, handler + formatter.
+
+def reset_logger(scope):
+    logger = logging.getLogger(scope)
+    logger.handlers = []
+    logger.setLevel(logging.NOTSET)
+    logger.propagate = True
+
+def test_setup_logging_w_no_scopes():
+    setup_logging()
+    base_logger = logging.getLogger("google")
+    assert base_logger.handlers == []
+    assert base_logger.propagate == False
+    assert base_logger.level == logging.NOTSET
+
+    reset_logger("google")
+
+
+def test_setup_logging_w_base_scope():
+    setup_logging("google")
+    base_logger = logging.getLogger("google")
+    assert isinstance(base_logger.handlers[0], logging.StreamHandler)
+    assert base_logger.propagate == False
+    assert base_logger.level == logging.DEBUG
+
+    reset_logger("google")
+
+def test_setup_logging_w_module_scope():
+    setup_logging("google.foo")
+
+    base_logger = logging.getLogger("google")
+    assert base_logger.handlers == []
+    assert base_logger.propagate == False
+    assert base_logger.level == logging.NOTSET
+
+    module_logger = logging.getLogger("google.foo")
+    assert isinstance(module_logger.handlers[0], logging.StreamHandler)
+    assert module_logger.propagate == False
+    assert module_logger.level == logging.DEBUG
+
+
+    reset_logger("google")
+
+def test_setup_logging_w_incorrect_scope():
+    setup_logging("foo")
+
+    base_logger = logging.getLogger("google")
+    assert base_logger.handlers == []
+    assert base_logger.propagate == False
+    assert base_logger.level == logging.NOTSET
+
+    # TODO: update test once we add logic to ignore an incorrect scope.
+    logger = logging.getLogger("foo")
+    assert isinstance(logger.handlers[0], logging.StreamHandler)
+    assert logger.propagate == False
+    assert logger.level == logging.DEBUG
+
+    reset_logger("google")