fba/models/error_log.py

   1 # Copyright (C) 2023 Free Software Foundation
   2 #
   3 # This program is free software: you can redistribute it and/or modify
   4 # it under the terms of the GNU Affero General Public License as published
   5 # by the Free Software Foundation, either version 3 of the License, or
   6 # (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU Affero General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU Affero General Public License
  14 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
  15
  16 import logging
  17 import json
  18 import time
  19
  20 from fba import database
  21
  22 from fba.helpers import blacklist
  23 from fba.helpers import config
  24 from fba.helpers import domain as domain_helper
  25
  26 logging.basicConfig(level=logging.INFO)
  27 logger = logging.getLogger(__name__)
  28
  29 def add(domain: str, error: dict):
  30     logger.debug("domain='%s',error[]='%s' - CALLED!", domain, type(error))
  31     domain_helper.raise_on(domain)
  32
  33     if blacklist.is_blacklisted(domain):
  34         raise ValueError(f"domain='{domain}' is blacklisted but function was invoked")
  35     elif config.get("write_error_log").lower() != "true":
  36         logger.debug("Writing to error_log is disabled in configuruation file - EXIT!")
  37         return
  38
  39     logger.debug("error[]='%s' - BEFORE!", type(error))
  40     if isinstance(error, (BaseException, json.decoder.JSONDecodeError)):
  41         error = f"error[{type(error)}]='{str(error)}'"
  42     logger.debug("error[]='%s' - AFTER!", type(error))
  43
  44     if isinstance(error, str):
  45         database.cursor.execute("INSERT INTO error_log (domain, error_code, error_message, created) VALUES (?, 999, ?, ?)",[
  46             domain,
  47             error,
  48             time.time()
  49         ])
  50     else:
  51         database.cursor.execute("INSERT INTO error_log (domain, error_code, error_message, created) VALUES (?, ?, ?, ?)",[
  52             domain,
  53             error["status_code"],
  54             error["error_message"],
  55             time.time()
  56         ])
  57
  58     # Cleanup old entries
  59     logger.debug("Purging old records (distance: %d)", config.get('error_log_cleanup'))
  60     database.cursor.execute("DELETE FROM error_log WHERE created < ?", [time.time() - config.get("error_log_cleanup")])
  61
  62     logger.debug("EXIT!")