feat: implement logger

src/firebase_functions/logger.py

@@ -0,0 +1,130 @@
+"""
+Logger module for Firebase Functions.
+"""
+
+import enum as _enum
+import json as _json
+import sys as _sys
+import typing as _typing
+import typing_extensions as _typing_extensions
+
+
+class LogSeverity(str, _enum.Enum):
+    """
+    `LogSeverity` indicates the detailed severity of the log entry. See
+    [LogSeverity](https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#logseverity).
+    """
+
+    DEBUG = "DEBUG"
+    INFO = "INFO"
+    NOTICE = "NOTICE"
+    WARNING = "WARNING"
+    ERROR = "ERROR"
+    CRITICAL = "CRITICAL"
+    ALERT = "ALERT"
+    EMERGENCY = "EMERGENCY"
+
+
+class LogEntry(_typing.TypedDict):
+    """
+    `LogEntry` represents a log entry.
+    See [LogEntry](https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry).
+    """
+
+    severity: _typing_extensions.Required[LogSeverity]
+    message: _typing_extensions.NotRequired[str]
+
+
+def _entry_from_args(severity: LogSeverity, *args, **kwargs) -> LogEntry:
+    """
+    Creates a `LogEntry` from the given arguments.
+    """
+
+    message: str = " ".join([
+        value
+        if isinstance(value, str) else _json.dumps(_remove_circular(value))
+        for value in args
+    ])
+
+    other: _typing.Dict[str, _typing.Any] = {
+        key: value if isinstance(value, str) else _remove_circular(value)
+        for key, value in kwargs.items()
+    }
+
+    entry: _typing.Dict[str, _typing.Any] = {"severity": severity, **other}
+    if message:
+        entry["message"] = message
+
+    return _typing.cast(LogEntry, entry)
+
+
+def _remove_circular(obj: _typing.Any,
+                     refs: _typing.Set[_typing.Any] | None = None):
+    """
+    Removes circular references from the given object and replaces them with "[CIRCULAR]".
+    """
+
+    if refs is None:
+        refs = set()
+
+    if id(obj) in refs:
+        return "[CIRCULAR]"
+
+    if not isinstance(obj, (str, int, float, bool, type(None))):
+        refs.add(id(obj))
+
+    if isinstance(obj, dict):
+        return {key: _remove_circular(value, refs) for key, value in obj.items()}
+    elif isinstance(obj, list):
+        return [_remove_circular(value, refs) for _, value in enumerate(obj)]
+    elif isinstance(obj, tuple):
+        return tuple(
+            _remove_circular(value, refs) for _, value in enumerate(obj))
+    else:
+        return obj
+
+
+def _get_write_file(severity: LogSeverity) -> _typing.TextIO:
+    if severity == LogSeverity.ERROR:
+        return _sys.stderr
+    return _sys.stdout
+
+
+def write(entry: LogEntry) -> None:
+    write_file = _get_write_file(entry["severity"])
+    print(_json.dumps(_remove_circular(entry)), file=write_file)
+
+
+def debug(*args, **kwargs) -> None:
+    """
+    Logs a debug message.
+    """
+    write(_entry_from_args(LogSeverity.DEBUG, *args, **kwargs))
+
+
+def log(*args, **kwargs) -> None:
+    """
+    Logs a log message.
+    """
+    write(_entry_from_args(LogSeverity.NOTICE, *args, **kwargs))
+
+
+def info(*args, **kwargs) -> None:
+    """
+    Logs an info message.
+    """
+    write(_entry_from_args(LogSeverity.INFO, *args, **kwargs))
+
+
+def warn(*args, **kwargs) -> None:
+    """
+    Logs a warning message.
+    """
+    write(_entry_from_args(LogSeverity.WARNING, *args, **kwargs))
+
+
+def error(*args, **kwargs) -> None:
+    """
+    Logs an error message.
+    """
+    write(_entry_from_args(LogSeverity.ERROR, *args, **kwargs))

src/firebase_functions/private/_identity_fn.py

@@ -114,7 +114,7 @@ def _auth_user_record_from_token_data(token_data: dict[str, _typing.Any]):
     return AuthUserRecord(
         uid=token_data["uid"],
         email=token_data.get("email"),
-        email_verified=token_data.get("email_verified"),
+        email_verified=bool(token_data.get("email_verified")),
         display_name=token_data.get("display_name"),
         photo_url=token_data.get("photo_url"),
         phone_number=token_data.get("phone_number"),

src/firebase_functions/private/util.py

@@ -387,6 +387,8 @@ def timestamp_conversion(time: str) -> _dt.datetime:
     elif precision_timestamp == PrecisionTimestamp.SECONDS:
         return second_timestamp_conversion(time)
 
+    raise ValueError("Invalid timestamp")
+
 
 def microsecond_timestamp_conversion(time: str) -> _dt.datetime:
     """Converts a microsecond timestamp and returns a datetime object of the current time in UTC"""

tests/test_logger.py

@@ -0,0 +1,81 @@
+"""
+Logger module tests.
+"""
+
+import pytest
+import json
+from firebase_functions import logger
+
+
+class TestLogger:
+    """
+    Tests for the logger module.
+    """
+
+    def test_format_should_be_valid_json(self,
+                                         capsys: pytest.CaptureFixture[str]):
+        logger.log(foo="bar")
+        raw_log_output = capsys.readouterr().out
+        try:
+            json.loads(raw_log_output)
+        except json.JSONDecodeError:
+            pytest.fail("Log output was not valid JSON.")
+
+    def test_log_should_have_severity(self, capsys: pytest.CaptureFixture[str]):
+        logger.log(foo="bar")
+        raw_log_output = capsys.readouterr().out
+        log_output = json.loads(raw_log_output)
+        assert "severity" in log_output
+
+    def test_severity_should_be_debug(self, capsys: pytest.CaptureFixture[str]):
+        logger.debug(foo="bar")
+        raw_log_output = capsys.readouterr().out
+        log_output = json.loads(raw_log_output)
+        assert log_output["severity"] == "DEBUG"
+
+    def test_severity_should_be_notice(self,
+                                       capsys: pytest.CaptureFixture[str]):
+        logger.log(foo="bar")
+        raw_log_output = capsys.readouterr().out
+        log_output = json.loads(raw_log_output)
+        assert log_output["severity"] == "NOTICE"
+
+    def test_severity_should_be_info(self, capsys: pytest.CaptureFixture[str]):
+        logger.info(foo="bar")
+        raw_log_output = capsys.readouterr().out
+        log_output = json.loads(raw_log_output)
+        assert log_output["severity"] == "INFO"
+
+    def test_severity_should_be_warning(self,
+                                        capsys: pytest.CaptureFixture[str]):
+        logger.warn(foo="bar")
+        raw_log_output = capsys.readouterr().out
+        log_output = json.loads(raw_log_output)
+        assert log_output["severity"] == "WARNING"
+
+    def test_severity_should_be_error(self, capsys: pytest.CaptureFixture[str]):
+        logger.error(foo="bar")
+        raw_log_output = capsys.readouterr().err
+        log_output = json.loads(raw_log_output)
+        assert log_output["severity"] == "ERROR"
+
+    def test_log_should_have_message(self, capsys: pytest.CaptureFixture[str]):
+        logger.log("bar")
+        raw_log_output = capsys.readouterr().out
+        log_output = json.loads(raw_log_output)
+        assert "message" in log_output
+
+    def test_log_should_have_other_keys(self,
+                                        capsys: pytest.CaptureFixture[str]):
+        logger.log(foo="bar")
+        raw_log_output = capsys.readouterr().out
+        log_output = json.loads(raw_log_output)
+        assert "foo" in log_output
+
+    def test_message_should_be_space_separated(
+            self, capsys: pytest.CaptureFixture[str]):
+        logger.log("bar", "qux")
+        expected_message = "bar qux"
+        raw_log_output = capsys.readouterr().out
+        log_output = json.loads(raw_log_output)
+        assert log_output["message"] == expected_message