fba/models/sources.py

   1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
   2 # Copyright (C) 2023 Free Software Foundation
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published
   6 # by the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
  16
  17 import logging
  18
  19 import time
  20
  21 from fba import database
  22
  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 fetch (source_domain: str) -> dict:
  30     logger.debug("source_domain='%s' - CALLED!", source_domain)
  31     domain_helper.raise_on(source_domain)
  32
  33     database.cursor.execute("SELECT * FROM sources WHERE source_domain = ? LIMIT 1", [source_domain])
  34
  35     row = database.cursor.fetchone()
  36     logger.debug("row[]='%s' - EXIT!", type(row))
  37     return row
  38
  39 def is_recent(source_domain: str) -> bool:
  40     logger.debug("source_domain='%s' - CALLED!", source_domain)
  41     domain_helper.raise_on(source_domain)
  42
  43     recent = False
  44
  45     row = fetch(source_domain)
  46     logger.debug("row[]='%s'", type(row))
  47     if row is not None:
  48         logger.debug("source_domain='%s',row[last_accessed]=%d", source_domain, row["last_accessed"])
  49         recent = (time.time() - row["last_accessed"]) <= config.get("source_last_access")
  50
  51     logger.debug("recent='%s' - EXIT!", recent)
  52     return recent
  53
  54 def update (source_domain: str):
  55     logger.debug("source_domain='%s' - CALLED!", source_domain)
  56     domain_helper.raise_on(source_domain)
  57
  58     row = fetch(source_domain)
  59     logger.debug("row[]='%s'", type(row))
  60     if row is None:
  61         # Add instance
  62         database.cursor.execute("INSERT INTO sources (source_domain, last_accessed) VALUES (?, ?)", [
  63             source_domain,
  64             time.time()
  65         ])
  66     else:
  67         # Update last_accessed
  68         database.cursor.execute("UPDATE sources SET last_accessed = ? WHERE source_domain = ? LIMIT 1", [
  69             time.time(),
  70             source_domain
  71         ])
  72
  73     logger.debug("Invoking commit() ...")
  74     database.connection.commit()
  75
  76     logger.debug("EXIT!")