-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[EDS-631] Prevents long queries from consuming duplicate resources if…
… queries are resubmitted by users (#137) * Optimize imports * Implement an object storage cache layer Provides an interface and two implementations. * Implement query synchronizer to allow processes to share a single query. * Update search interface to take advantage of new query synchronizer to attach to ongoing queries and return their results. * [Bot] Update version to 2.2.0 Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>
- Loading branch information
1 parent
62f2ad0
commit b4280d0
Showing
12 changed files
with
551 additions
and
141 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,122 @@ | ||
| import json | ||
| import time | ||
| from abc import ABC, abstractmethod | ||
| from copy import copy | ||
| from enum import Enum | ||
| from typing import Any, Optional | ||
|
|
||
| from flask_injector import request | ||
| from injector import Module, provider | ||
| from pydantic import BaseModel | ||
| from redis import Redis | ||
|
|
||
| from husky_directory.app_config import ApplicationConfig | ||
| from husky_directory.util import AppLoggerMixIn | ||
|
|
||
|
|
||
| class ObjectStorageInterface(AppLoggerMixIn, ABC): | ||
| """ | ||
| Basic interface that does nothing but declare | ||
| abstractions. | ||
| It also provides a utility method that can convert | ||
| anything* into a string. | ||
| *if the thing you want to convert can't be converted, | ||
| add a case in the normalize_object_data implementation below. | ||
| """ | ||
|
|
||
| @abstractmethod | ||
| def get(self, key: str) -> Optional[str]: | ||
| return None | ||
|
|
||
| @abstractmethod | ||
| def put(self, key: str, obj: Any, expire_after_seconds: Optional[int] = None): | ||
| pass | ||
|
|
||
| @staticmethod | ||
| def normalize_object_data(obj: Any) -> str: | ||
| if isinstance(obj, BaseModel): | ||
| return obj.json() | ||
| if isinstance(obj, Enum): | ||
| obj = obj.value | ||
| if not isinstance(obj, str): | ||
| return json.dumps(obj) | ||
| return obj | ||
|
|
||
|
|
||
| class InMemoryObjectStorage(ObjectStorageInterface): | ||
| """ | ||
| Used when testing locally using flask itself, | ||
| cannot be shared between processes. This is a very | ||
| basic implementation which checks for key expiration | ||
| on every `put`. | ||
| """ | ||
|
|
||
| def __init__(self): | ||
| self.__store__ = {} | ||
| self.__key_expirations__ = {} | ||
|
|
||
| def validate_key_expiration(self, key: str): | ||
| expiration = self.__key_expirations__.get(key) | ||
| now = time.time() | ||
| if expiration: | ||
| max_elapsed = expiration["max"] | ||
| if not max_elapsed: | ||
| return | ||
| elapsed = now - expiration["stored"] | ||
| if elapsed > max_elapsed: | ||
| del self.__key_expirations__[key] | ||
| if key in self.__store__: | ||
| del self.__store__[key] | ||
|
|
||
| def expire_keys(self): | ||
| for key in copy(self.__key_expirations__): | ||
| self.validate_key_expiration(key) | ||
|
|
||
| def get(self, key: str) -> Optional[str]: | ||
| self.validate_key_expiration(key) | ||
| return self.__store__.get(key) | ||
|
|
||
| def put(self, key: str, obj: Any, expire_after_seconds: Optional[int] = None): | ||
| self.expire_keys() | ||
| self.__store__[key] = self.normalize_object_data(obj) | ||
| now = time.time() | ||
| self.__key_expirations__[key] = {"stored": now, "max": expire_after_seconds} | ||
| return key | ||
|
|
||
|
|
||
| class RedisObjectStorage(ObjectStorageInterface): | ||
| def __init__(self, redis: Redis, config: ApplicationConfig): | ||
| self.redis = redis | ||
| self.prefix = f"{config.redis_settings.namespace}:obj" | ||
|
|
||
| def normalize_key(self, key: str) -> str: | ||
| """Normalizes the key using the configured namespace.""" | ||
| if not key.startswith(self.prefix): | ||
| key = f"{self.prefix}:{key}" | ||
| return key | ||
|
|
||
| def put(self, key: str, obj: Any, expire_after_seconds: Optional[int] = None): | ||
| key = self.normalize_key(key) | ||
| self.redis.set(key, self.normalize_object_data(obj), ex=expire_after_seconds) | ||
| return key | ||
|
|
||
| def get(self, key: str) -> Optional[str]: | ||
| val = self.redis.get(self.normalize_key(key)) | ||
| if val: | ||
| if isinstance(val, bytes): | ||
| return val.decode("UTF-8") | ||
| return val | ||
| return None | ||
|
|
||
|
|
||
| class ObjectStoreInjectorModule(Module): | ||
| @request | ||
| @provider | ||
| def provide_object_store( | ||
| self, redis: Redis, config: ApplicationConfig | ||
| ) -> ObjectStorageInterface: | ||
| if config.redis_settings.host: | ||
| return RedisObjectStorage(redis, config) | ||
| return InMemoryObjectStorage() |
Oops, something went wrong.