]> git.mxchange.org Git - fba.git/blob - fba/helpers/cache.py
Continued:
[fba.git] / fba / helpers / cache.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 import sqlite3
19
20 logging.basicConfig(level=logging.INFO)
21 logger = logging.getLogger(__name__)
22
23 # Cache for redundant SQL queries
24 _cache = {}
25
26 ##### Cache #####
27
28 def key_exists(key: str) -> bool:
29     logger.debug("key='%s' - CALLED!", key)
30     exists = key in _cache
31
32     logger.debug("exists='%s' - EXIT!", exists)
33     return exists
34
35 def set_all(key: str, rows: list, value: any):
36     logger.debug("key='%s',rows()=%d,value[]='%s' - CALLED!", key, len(rows), type(value))
37
38     if not isinstance(key, str):
39         raise ValueError(f"Parameter key[]='{type(key)}' is not of type 'str'")
40     elif not key_exists(key):
41         logger.debug("Cache for key='%s' not initialized.", key)
42         _cache[key] = dict()
43
44     logger.debug("Setting %d row(s) for key='%s',value[%s]='%s' ...", len(rows), key, type(value), value)
45     for sub in rows:
46         logger.debug("Setting key='%s',sub[%s]='%s'", key, type(sub), sub)
47         if isinstance(sub, sqlite3.Row):
48             logger.debug("Setting key='%s',sub[%s]='%s',value[]='%s'", key, type(sub), sub, type(value))
49             _cache[key][sub[0]] = value
50         else:
51             logger.warning("Unsupported type sub[]='%s'", type(sub))
52
53     logger.debug("EXIT!")
54
55 def set_sub_key(key: str, sub: str, value: any):
56     logger.debug("key='%s',sub='%s',value[]='%s' - CALLED!", key, sub, type(value))
57
58     if not isinstance(key, str):
59         raise ValueError(f"Parameter key[]='{type(key)}' is not of type 'str'")
60     elif not isinstance(sub, str):
61         raise ValueError(f"Parameter sub[]='{type(sub)}' is not of type 'str'")
62     elif not key_exists(key):
63         raise Exception(f"Cache for key='{key}' is not initialized, but function invoked")
64
65     logger.debug("Setting key='%s',sub='%s',value[%s]='%s' ...", key, sub, type(value), value)
66     _cache[key][sub] = value
67
68     logger.debug("EXIT!")
69
70 def sub_key_exists(key: str, sub: str) -> bool:
71     logger.debug("key='%s',sub='%s' - CALLED!", key, sub)
72
73     if not isinstance(key, str):
74         raise ValueError(f"Parameter key[]='{type(key)}' is not of type 'str'")
75     elif not isinstance(sub, str):
76         raise ValueError(f"Parameter sub[]='{type(sub)}' is not of type 'str'")
77     elif not key_exists(key):
78         raise Exception(f"Cache for key='{key}' is not initialized, but function invoked")
79
80     exists = sub in _cache[key]
81
82     logger.debug("exists='%s' - EXIT!", exists)
83     return exists