Egor fureunoir Gorbunov
a81f734e23
Fixes: (1) Removed all `# type: ignore` annotations across the codebase; (2) Fixed usage of Django Model methods by eliminating unnecessary `# type: ignore` directives; (3) Adjusted usage of functions like `get()` to align with method expectations, removing incorrect comments; Extra: (1) Deleted `pyrightconfig.json` as part of migration to a stricter type-checked environment; (2) Minor code cleanup, including formatting changes and refactoring import statements in adherence to PEP8 recommendations.
62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, Type
|
|
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
from django.core.exceptions import BadRequest
|
|
from django.utils.translation import gettext_lazy as _
|
|
from graphene import Context
|
|
from rest_framework.request import Request
|
|
|
|
from engine.vibes_auth.models import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def is_safe_cache_key(key: str) -> bool:
|
|
return key not in settings.UNSAFE_CACHE_KEYS
|
|
|
|
|
|
def get_cached_value(user: Type[User], key: str, default: Any = None) -> Any:
|
|
if user.is_staff or user.is_superuser:
|
|
return cache.get(key, default)
|
|
|
|
if is_safe_cache_key(key):
|
|
return cache.get(key, default)
|
|
|
|
return None
|
|
|
|
|
|
def set_cached_value(
|
|
user: Type[User], key: str, value: object, timeout: int = 3600
|
|
) -> None | object:
|
|
if user.is_staff or user.is_superuser:
|
|
cache.set(key, value, timeout)
|
|
return value
|
|
|
|
return None
|
|
|
|
|
|
def web_cache(
|
|
request: Request | Context, key: str, data: dict[str, Any], timeout: int
|
|
) -> dict[str, Any]:
|
|
if not data and not timeout:
|
|
return {"data": get_cached_value(request.user, key)}
|
|
if (data and not timeout) or (timeout and not data):
|
|
raise BadRequest(_("both data and timeout are required"))
|
|
if not 0 < int(timeout) < 216000:
|
|
raise BadRequest(
|
|
_("invalid timeout value, it must be between 0 and 216000 seconds")
|
|
)
|
|
return {"data": set_cached_value(request.user, key, data, timeout)}
|
|
|
|
|
|
def set_default_cache() -> None:
|
|
data_dir = Path(__file__).resolve().parent.parent / "data"
|
|
for json_file in data_dir.glob("*.json"):
|
|
with json_file.open("r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
logger.info("Setting cache for %s", json_file.stem)
|
|
cache.set(json_file.stem, data, timeout=28800)
|