Introduces convenient logging functions for usage in loops:

The functions with the postfix `_once` only log when they are called for the first time. This is useful when used inside loops while the arguments to be logged do not change. The current alternatives (`logging.info`, `logging.warning`, etc.) can load the log too heavily here.

* `log_once(level, msg, *args)` Logs only the first call. (Forwards to `log_first_n`)
* `debug_once(msg, *args)` forwards to `log_once`.
* `info_once(msg, *args)` forwards to `log_once`.
* `warning_once(msg, *args)` forwards to `log_once`.
* `error_once(msg, *args)` forwards to `log_once`.
* `fatal_once(msg, *args)` forwards to `log_once`.

PiperOrigin-RevId: 535561745

absl/logging/__init__.py

@@ -487,6 +487,45 @@ def log_every_n(level, msg, n, *args):
   log_if(level, msg, not (count % n), *args)
 
 
+def log_once(level, msg, *args):
+  """Logs ``msg % args`` at lebel 'lebel' once.
+
+  Logs only the first call.
+  Not threadsafe.
+
+  Args:
+    level: int, the absl logging level at which to log.
+    msg: str, the message to be logged.
+    *args: The args to be substituted into the msg.
+  """
+  log_first_n(level, msg, 1, *args)
+
+
+def fatal_once(msg, *args):
+  """Logs a fatal message once."""
+  log_once(FATAL, msg, *args)
+
+
+def error_once(msg, *args):
+  """Logs an error message once."""
+  log_once(ERROR, msg, *args)
+
+
+def warning_once(msg, *args):
+  """Logs a warning message once."""
+  log_once(WARNING, msg, *args)
+
+
+def info_once(msg, *args):
+  """Logs an info message once."""
+  log_once(INFO, msg, *args)
+
+
+def debug_once(msg, *args):
+  """Logs a debug message once."""
+  log_once(DEBUG, msg, *args)
+
+
 # Keeps track of the last log time of the given token.
 # Note: must be a dict since set/get is atomic in CPython.
 # Note: entries are never released as their number is expected to be low.

absl/logging/tests/logging_test.py

@@ -688,7 +688,7 @@ def t():
     self.assertEqual(counts, {i for i in range(100)})
 
 
-class LoggingTest(absltest.TestCase):
+class LoggingTest(parameterized.TestCase):
 
   def test_fatal(self):
     with mock.patch.object(os, 'abort') as mock_abort:
@@ -901,6 +901,38 @@ def test_set_verbosity_strings(self):
 
     logging.set_verbosity(old_level)
 
+  @parameterized.parameters([
+      dict(logging_fn=logging.error_once),
+      dict(logging_fn=logging.warning_once),
+      dict(logging_fn=logging.info_once),
+      dict(logging_fn=functools.partial(logging.log_once, logging.INFO)),
+      dict(logging_fn=logging.debug_once),
+  ])
+  def test_log_once(self, logging_fn):
+    logging._log_counter_per_token = {}
+    FLAGS.verbosity = 1  # Needed for testing debug_once.
+    with mock.patch.object(sys, 'stderr', new=io.StringIO()) as stderr:
+      for i in range(5):
+        logging_fn('iteration %d', i)
+      logged_lines = stderr.getvalue().strip().split('\n')
+      self.assertLen(logged_lines, 1)
+      self.assertRegex(logged_lines[0], 'iteration 0')
+      self.assertNotRegex(logged_lines[0], 'iteration 1')
+
+  def test_fatal_once(self):
+    with (
+        mock.patch.object(sys, 'stderr', new=io.StringIO()) as stderr,
+        mock.patch.object(os, 'abort') as mock_abort,
+    ):
+      for i in range(5):
+        logging.fatal_once('iteration %d', i)
+        mock_abort.assert_called_once()
+
+        logged_lines = stderr.getvalue().strip().split('\n')
+        self.assertLen(logged_lines, 1)
+        self.assertRegex(logged_lines[0], 'iteration 0')
+        self.assertNotRegex(logged_lines[0], 'iteration 1')
+
   def test_key_flags(self):
     key_flags = FLAGS.get_key_flags_for_module(logging)
     key_flag_names = [flag.name for flag in key_flags]