Added experimental cache support
All checks were successful
Analysis / SonarCloud (push) Successful in 51s
All checks were successful
Analysis / SonarCloud (push) Successful in 51s
This commit is contained in:
1
src/libbot/cache/utils/__init__.py
vendored
Normal file
1
src/libbot/cache/utils/__init__.py
vendored
Normal file
@@ -0,0 +1 @@
|
||||
from .manager import create_cache_client
|
42
src/libbot/cache/utils/_objects.py
vendored
Normal file
42
src/libbot/cache/utils/_objects.py
vendored
Normal file
@@ -0,0 +1,42 @@
|
||||
import logging
|
||||
from copy import deepcopy
|
||||
from logging import Logger
|
||||
from typing import Any
|
||||
|
||||
try:
|
||||
from ujson import dumps, loads
|
||||
except ImportError:
|
||||
from json import dumps, loads
|
||||
|
||||
logger: Logger = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
from bson import ObjectId
|
||||
except ImportError:
|
||||
logger.warning(
|
||||
"Could not import bson.ObjectId. PyMongo conversions will not be supported by the cache. It's safe to ignore this message if you do not use MongoDB."
|
||||
)
|
||||
|
||||
|
||||
def _json_to_string(json_object: Any) -> str:
|
||||
json_object_copy: Any = deepcopy(json_object)
|
||||
|
||||
if isinstance(json_object_copy, dict) and "_id" in json_object_copy:
|
||||
json_object_copy["_id"] = str(json_object_copy["_id"])
|
||||
|
||||
return dumps(json_object_copy, ensure_ascii=False, indent=0, escape_forward_slashes=False)
|
||||
|
||||
|
||||
def _string_to_json(json_string: str) -> Any:
|
||||
json_object: Any = loads(json_string)
|
||||
|
||||
if "_id" in json_object:
|
||||
try:
|
||||
json_object["_id"] = ObjectId(json_object["_id"])
|
||||
except NameError:
|
||||
logger.debug(
|
||||
"Tried to convert attribute '_id' with value '%s' but bson.ObjectId is not present, skipping the conversion.",
|
||||
json_object["_id"],
|
||||
)
|
||||
|
||||
return json_object
|
24
src/libbot/cache/utils/manager.py
vendored
Normal file
24
src/libbot/cache/utils/manager.py
vendored
Normal file
@@ -0,0 +1,24 @@
|
||||
from typing import Dict, Any, Literal
|
||||
|
||||
from ..classes import CacheMemcached, CacheRedis
|
||||
|
||||
|
||||
def create_cache_client(
|
||||
config: Dict[str, Any],
|
||||
engine: Literal["memcached", "redis"] | None = None,
|
||||
) -> CacheMemcached | CacheRedis:
|
||||
if engine not in ["memcached", "redis"] or engine is None:
|
||||
raise KeyError(f"Incorrect cache engine provided. Expected 'memcached' or 'redis', got '{engine}'")
|
||||
|
||||
if "cache" not in config or engine not in config["cache"]:
|
||||
raise KeyError(
|
||||
f"Cache configuration is invalid. Please check if all keys are set (engine: '{engine}')"
|
||||
)
|
||||
|
||||
match engine:
|
||||
case "memcached":
|
||||
return CacheMemcached.from_config(config["cache"][engine])
|
||||
case "redis":
|
||||
return CacheRedis.from_config(config["cache"][engine])
|
||||
case _:
|
||||
raise KeyError(f"Cache implementation for the engine '{engine}' is not present.")
|
Reference in New Issue
Block a user