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/>.
19 logging.basicConfig(level=logging.INFO)
20 logger = logging.getLogger(__name__)
22 # Cache for redundant SQL queries
27 def key_exists(key: str) -> bool:
30 def set_all(key: str, rows: list, value: any):
31 logger.debug("key='%s',rows()=%d,value[]='%s' - CALLED!", key, len(rows), type(value))
32 if not isinstance(key, str):
33 raise ValueError(f"Parameter key[]='{type(key)}' is not 'str'")
34 elif not key_exists(key):
35 logger.debug("Cache for key='%s' not initialized.", key)
39 logger.debug("Setting key='%s',sub[%s]='%s'", key, type(sub), sub)
40 if isinstance(sub, tuple):
41 logger.debug("Setting key='%s',sub[%s]='%s',value[]='%s'", key, type(sub), sub, type(value))
42 _cache[key][sub[0]] = value
44 logger.warning("Unsupported type sub[]='%s'", type(sub))
48 def set_sub_key(key: str, sub: str, value: any):
49 logger.debug("key='%s',sub='%s',value[]='%s' - CALLED!", key, sub, type(value))
50 if not isinstance(key, str):
51 raise ValueError(f"Parameter key[]='{type(key)}' is not 'str'")
52 elif not isinstance(sub, str):
53 raise ValueError(f"Parameter sub[]='{type(sub)}' is not 'str'")
54 elif not key_exists(key):
55 raise Exception(f"Cache for key='{key}' is not initialized, but function invoked")
57 logger.debug("Setting key='%s',sub='%s',value[]='%s' ...", key, sub, type(value))
58 _cache[key][sub] = value
62 def sub_key_exists(key: str, sub: str) -> bool:
63 logger.debug("key='%s',sub='%s' - CALLED!", key, sub)
64 if not isinstance(key, str):
65 raise ValueError(f"Parameter key[]='{type(key)}' is not 'str'")
66 elif not isinstance(sub, str):
67 raise ValueError(f"Parameter sub[]='{type(sub)}' is not 'str'")
68 elif not key_exists(key):
69 raise Exception(f"Cache for key='{key}' is not initialized, but function invoked")
71 exists = sub in _cache[key]
73 logger.debug("exists='%s' - EXIT!", exists)