1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
2 # Copyright (C) 2023 Free Software Foundation
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.
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.
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/>.
21 from fba import database
23 from fba.helpers import config
24 from fba.helpers import domain as domain_helper
26 logging.basicConfig(level=logging.INFO)
27 logger = logging.getLogger(__name__)
29 def fetch (source_domain: str) -> dict:
30 logger.debug("source_domain='%s' - CALLED!", source_domain)
31 domain_helper.raise_on(source_domain)
33 database.cursor.execute("SELECT * FROM sources WHERE source_domain = ? LIMIT 1", [source_domain])
35 row = database.cursor.fetchone()
36 logger.debug("row[]='%s' - EXIT!", type(row))
39 def is_recent(source_domain: str) -> bool:
40 logger.debug("source_domain='%s' - CALLED!", source_domain)
41 domain_helper.raise_on(source_domain)
45 row = fetch(source_domain)
46 logger.debug("row[]='%s'", type(row))
48 logger.debug("source_domain='%s',row[last_accessed]=%d", source_domain, row["last_accessed"])
49 is_recent = (time.time() - row["last_accessed"]) <= config.get("source_last_access")
51 logger.debug("is_recent='%s' - EXIT!", is_recent)
54 def update (source_domain: str):
55 logger.debug("source_domain='%s' - CALLED!", source_domain)
56 domain_helper.raise_on(source_domain)
58 row = fetch(source_domain)
59 logger.debug("row[]='%s'", type(row))
62 database.cursor.execute("INSERT INTO sources (source_domain, last_accessed) VALUES (?, ?)", [
67 # Update last_accessed
68 database.cursor.execute("UPDATE sources SET last_accessed = ? WHERE source_domain = ? LIMIT 1", [
73 logger.debug("Invoking commit() ...")
74 database.connection.commit()