NextcloudApp: setup_nextcloud_logging function for transparent logging (#294)

Usage:

```python3
import logging
from pathlib import Path
from os import environ

import nc_py_api


environ["APP_ID"] = "nc_py_api"
environ["APP_VERSION"] = "1.0.0"
environ["APP_SECRET"] = "12345"
environ["NEXTCLOUD_URL"] = "http://nextcloud.local"


if __name__ == "__main__":
    nc_app = nc_py_api.NextcloudApp()
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s: [%(funcName)s]:%(levelname)s: %(message)s",
        datefmt="%H:%M:%S",
    )
    logging.getLogger("httpx").setLevel(logging.ERROR)  # not needed, but better to hide spam to console
    nc_py_api.ex_app.setup_nextcloud_logging() # setup logging handler in one line of code
    logging.fatal("Fatal test")
    logging.error("Error test")
    logging.warning("Warning test")
    logging.info("Info test")
    logging.debug("Debug test")
    logging.fatal("Fatal test2")
    try:
        a = 0
        b = z
    except Exception as e:
        logging.exception("Exception test")


```

### setup_nextcloud_logging

```python3 

def setup_nextcloud_logging(logger_name: str | None = None, logging_level: int = logging.DEBUG):
     """Function to easily send all or selected log entries to Nextcloud."""
     logger = logging.getLogger(logger_name)
     nextcloud_handler = _NextcloudStorageHandler()
     nextcloud_handler.setLevel(logging_level)
     logger.addHandler(nextcloud_handler)
     return nextcloud_handler
```

---------

Signed-off-by: Alexander Piskun <[email protected]>

Author: bigcat88

CHANGELOG.md

@@ -4,9 +4,13 @@ All notable changes to this project will be documented in this file.
 
 ## [0.17.1 - 2024-09-06]
 
+### Added
+
+- NextcloudApp: `setup_nextcloud_logging` function to support transparently sending logs to Nextcloud. #294
+
 ### Fixed
 
-- NextcloudApp: `nc.log` now suppresses all exceptions to safe call it anywhere in your app.
+- NextcloudApp: `nc.log` now suppresses all exceptions to safe call it anywhere in your app. #293
 
 ## [0.17.0 - 2024-09-05]
 

nc_py_api/ex_app/__init__.py

@@ -10,6 +10,7 @@
     set_handlers,
     talk_bot_msg,
 )
+from .logging import setup_nextcloud_logging
 from .misc import (
     get_computation_device,
     get_model_path,

nc_py_api/ex_app/logging.py

@@ -0,0 +1,46 @@
+"""Transparent logging support to store logs in the nextcloud.log."""
+
+import logging
+import threading
+
+from ..nextcloud import NextcloudApp
+from .defs import LogLvl
+
+LOGLVL_MAP = {
+    logging.NOTSET: LogLvl.DEBUG,
+    logging.DEBUG: LogLvl.DEBUG,
+    logging.INFO: LogLvl.INFO,
+    logging.WARNING: LogLvl.WARNING,
+    logging.ERROR: LogLvl.ERROR,
+    logging.CRITICAL: LogLvl.FATAL,
+}
+
+THREAD_LOCAL = threading.local()
+
+
+class _NextcloudLogsHandler(logging.Handler):
+    def __init__(self):
+        super().__init__()
+
+    def emit(self, record):
+        if THREAD_LOCAL.__dict__.get("nc_py_api.loghandler", False):
+            return
+
+        try:
+            THREAD_LOCAL.__dict__["nc_py_api.loghandler"] = True
+            log_entry = self.format(record)
+            log_level = record.levelno
+            NextcloudApp().log(LOGLVL_MAP.get(log_level, LogLvl.FATAL), log_entry, fast_send=True)
+        except Exception:  # noqa pylint: disable=broad-exception-caught
+            self.handleError(record)
+        finally:
+            THREAD_LOCAL.__dict__["nc_py_api.loghandler"] = False
+
+
+def setup_nextcloud_logging(logger_name: str | None = None, logging_level: int = logging.DEBUG):
+    """Function to easily send all or selected log entries to Nextcloud."""
+    logger = logging.getLogger(logger_name)
+    nextcloud_handler = _NextcloudLogsHandler()
+    nextcloud_handler.setLevel(logging_level)
+    logger.addHandler(nextcloud_handler)
+    return nextcloud_handler

nc_py_api/nextcloud.py

@@ -348,15 +348,16 @@ def enabled_state(self) -> bool:
             return bool(self._session.ocs("GET", "/ocs/v1.php/apps/app_api/ex-app/state"))
         return False
 
-    def log(self, log_lvl: LogLvl, content: str) -> None:
+    def log(self, log_lvl: LogLvl, content: str, fast_send: bool = False) -> None:
         """Writes log to the Nextcloud log file."""
-        if self.check_capabilities("app_api"):
-            return
         int_log_lvl = int(log_lvl)
         if int_log_lvl < 0 or int_log_lvl > 4:
             raise ValueError("Invalid `log_lvl` value")
-        if int_log_lvl < self.capabilities["app_api"].get("loglevel", 0):
-            return
+        if not fast_send:
+            if self.check_capabilities("app_api"):
+                return
+            if int_log_lvl < self.capabilities["app_api"].get("loglevel", 0):
+                return
         with contextlib.suppress(Exception):
             self._session.ocs("POST", f"{self._session.ae_url}/log", json={"level": int_log_lvl, "message": content})
 
@@ -482,15 +483,16 @@ async def enabled_state(self) -> bool:
             return bool(await self._session.ocs("GET", "/ocs/v1.php/apps/app_api/ex-app/state"))
         return False
 
-    async def log(self, log_lvl: LogLvl, content: str) -> None:
+    async def log(self, log_lvl: LogLvl, content: str, fast_send: bool = False) -> None:
         """Writes log to the Nextcloud log file."""
-        if await self.check_capabilities("app_api"):
-            return
         int_log_lvl = int(log_lvl)
         if int_log_lvl < 0 or int_log_lvl > 4:
             raise ValueError("Invalid `log_lvl` value")
-        if int_log_lvl < (await self.capabilities)["app_api"].get("loglevel", 0):
-            return
+        if not fast_send:
+            if await self.check_capabilities("app_api"):
+                return
+            if int_log_lvl < (await self.capabilities)["app_api"].get("loglevel", 0):
+                return
         with contextlib.suppress(Exception):
             await self._session.ocs(
                 "POST", f"{self._session.ae_url}/log", json={"level": int_log_lvl, "message": content}

tests/actual_tests/logs_test.py

@@ -1,9 +1,10 @@
+import logging
 from copy import deepcopy
 from unittest import mock
 
 import pytest
 
-from nc_py_api.ex_app import LogLvl
+from nc_py_api.ex_app import LogLvl, setup_nextcloud_logging
 
 
 def test_loglvl_values():
@@ -113,3 +114,23 @@ async def test_log_without_app_api_async(anc_app):
     ):
         await anc_app.log(log_lvl, "will not be sent")
         ocs.assert_not_called()
+
+
+def test_logging(nc_app):
+    log_handler = setup_nextcloud_logging("my_logger")
+    logger = logging.getLogger("my_logger")
+    logger.fatal("testing logging.fatal")
+    try:
+        a = b  # noqa
+    except Exception:  # noqa
+        logger.exception("testing logger.exception")
+    logger.removeHandler(log_handler)
+
+
+def test_recursive_logging(nc_app):
+    logging.getLogger("httpx").setLevel(logging.DEBUG)
+    log_handler = setup_nextcloud_logging()
+    logger = logging.getLogger()
+    logger.fatal("testing logging.fatal")
+    logger.removeHandler(log_handler)
+    logging.getLogger("httpx").setLevel(logging.ERROR)