diff options
3 files changed, 29 insertions, 65 deletions
diff --git a/generalresearch/managers/thl/user_manager/memcached_user_manager.py b/generalresearch/managers/thl/user_manager/memcached_user_manager.py deleted file mode 100644 index d2c68ed..0000000 --- a/generalresearch/managers/thl/user_manager/memcached_user_manager.py +++ /dev/null @@ -1,49 +0,0 @@ -# from typing import List, Optional -# -# import pylibmc -# -# from generalresearch.models.thl.user import User -# -# -# class MemcachedUserManager: -# def __init__(self, servers: List[str], cache_prefix: Optional[str] = None): -# self.servers = servers -# self.cache_prefix = cache_prefix if cache_prefix else "user-lookup" -# -# def create_client(self): -# # Clients are NOT thread safe. Make a new one each time -# -# # There's a receive_timeout and send_timeout also, but the documentation is incomprehensible, -# # and they don't seem to do anything??? (I tested setting them at 1ms and I can't -# # get it to fail) -# # https://sendapatch.se/projects/pylibmc/behaviors.html -# mc_client = pylibmc.Client(servers=self.servers, binary=True, -# behaviors={'connect_timeout': 100}) -# return mc_client -# -# def get_user(self, *, product_id: str = None, product_user_id: str = None, user_id: int = None, -# user_uuid: UUIDStr = None) -> User: -# # assume we did input validation in user_manager.get_user() function -# mc_client = self.create_client() -# if user_uuid: -# d = mc_client.get(f"{self.cache_prefix}:uuid:{user_uuid}") -# elif user_id: -# d = mc_client.get(f"{self.cache_prefix}:user_id:{user_id}") -# else: -# d = mc_client.get(f"{self.cache_prefix}:ubp:{product_id}:{product_user_id}") -# if d: -# return User.model_validate_json(d) -# -# def set_user(self, user: User): -# d = user.to_json() -# mc_client = self.create_client() -# mc_client.set(f"{self.cache_prefix}:uuid:{user.uuid}", d, time=60 * 60 * 24) -# mc_client.set(f"{self.cache_prefix}:user_id:{user.user_id}", d, time=60 * 60 * 24) -# mc_client.set(f"{self.cache_prefix}:ubp:{user.product_id}:{user.product_user_id}", d, time=60 * 60 * 24) -# -# def clear_user(self, user: User): -# # this should only be used by tests -# mc_client = self.create_client() -# mc_client.delete(f"{self.cache_prefix}:uuid:{user.uuid}") -# mc_client.delete(f"{self.cache_prefix}:user_id:{user.user_id}") -# mc_client.delete(f"{self.cache_prefix}:ubp:{user.product_id}:{user.product_user_id}") diff --git a/generalresearch/managers/thl/user_manager/mysql_user_manager.py b/generalresearch/managers/thl/user_manager/mysql_user_manager.py index dbed5de..22abaa8 100644 --- a/generalresearch/managers/thl/user_manager/mysql_user_manager.py +++ b/generalresearch/managers/thl/user_manager/mysql_user_manager.py @@ -1,11 +1,13 @@ from __future__ import annotations import logging +import operator from collections.abc import Collection from datetime import datetime, timezone -from functools import lru_cache +from threading import Lock from uuid import uuid4 +from cachetools import LRUCache, cachedmethod import psycopg from psycopg import sql @@ -22,6 +24,8 @@ class MysqlUserManager: def __init__(self, pg_config: PostgresConfig, is_read_replica: bool): self.pg_config = pg_config self.is_read_replica = is_read_replica + self.product_id_exists_cache = LRUCache(maxsize=5000) + self.product_id_exists_cache_lock = Lock() def _set_last_seen(self, user: User) -> None: # Don't call this directly. Use UserManager.set_last_seen() @@ -143,7 +147,7 @@ class MysqlUserManager: with conn.cursor() as c: c.execute(query=query, params=params) user_id = c.fetchone()["id"] - except psycopg.IntegrityError as e: + except psycopg.IntegrityError: # Two machines/processes are trying to create this same (product_id, product_user_id) # at the same time. There's a unique index, so mysql will not let two be created. # The 2nd should get an IntegrityError, meaning this already exists, and we can just query it. @@ -160,7 +164,7 @@ class MysqlUserManager: else: # We specifically queried the NON read-replica, and we got an IntegrityError, so # something else must be wrong... - raise e + raise else: user = User( user_id=user_id, @@ -173,8 +177,11 @@ class MysqlUserManager: return user - @lru_cache(maxsize=5000) - def product_id_exists(self, product_id: str): + @cachedmethod( + operator.attrgetter("product_id_exists_cache"), + lock=operator.attrgetter("product_id_exists_cache_lock"), + ) + def product_id_exists(self, product_id: str) -> bool: # 'id' is the primary key, there can only be 0 or 1 query = """ SELECT id @@ -254,10 +261,10 @@ class MysqlUserManager: user_ids and user_uuids ), "Must pass ONE of user_ids, user_uuids" if user_ids: - assert len(user_ids) <= 500, "limit 500 user_ids" assert isinstance( user_ids, (list, set) ), "must pass a collection of user_ids" + assert len(user_ids) <= 500, "limit 500 user_ids" res = self.pg_config.execute_sql_query( query=""" @@ -270,10 +277,10 @@ class MysqlUserManager: params={"user_ids": user_ids}, ) else: - assert len(user_uuids) <= 500, "limit 500 user_uuids" assert isinstance( user_uuids, (list, set) ), "must pass a collection of user_uuids" + assert len(user_uuids) <= 500, "limit 500 user_uuids" res = self.pg_config.execute_sql_query( query=""" SELECT id AS user_id, product_id, product_user_id, diff --git a/generalresearch/managers/thl/user_manager/user_manager.py b/generalresearch/managers/thl/user_manager/user_manager.py index 3794020..5da3370 100644 --- a/generalresearch/managers/thl/user_manager/user_manager.py +++ b/generalresearch/managers/thl/user_manager/user_manager.py @@ -1,10 +1,12 @@ from __future__ import annotations import logging +import operator from collections.abc import Collection from datetime import datetime -from functools import lru_cache +from threading import Lock +from cachetools import TTLCache, cachedmethod from pydantic import RedisDsn from generalresearch.managers.base import Permission @@ -78,6 +80,8 @@ class UserManager: self.product_manager = ProductManager( pg_config=pg_config, permissions=[Permission.READ] ) + self.get_user_cache = TTLCache(maxsize=10000, ttl=30) + self.get_user_cache_lock = Lock() def set_last_seen(self, user: User) -> None: assert Permission.UPDATE in self.sql_permissions, "permission error" @@ -105,14 +109,17 @@ class UserManager: return None - def cache_clear(self): - # Generally this is used in testing. This clears the .get_user's lru_cache. - # There is no way of clearing only a specific key from the cache. + def cache_clear(self) -> None: + # Generally this is used in testing. This clears get_user's TTL cache. # It does not clear any redis caches; that has to be done separately. - self.get_user.__wrapped__.cache_clear() + with self.get_user_cache_lock: + self.get_user_cache.clear() @deepcopy_return - @lru_cache(maxsize=10000) + @cachedmethod( + operator.attrgetter("get_user_cache"), + lock=operator.attrgetter("get_user_cache_lock"), + ) def get_user( self, *, @@ -123,7 +130,7 @@ class UserManager: ) -> User: """ Retrieve User from (product_id & product_user_id) or (user_id), or (uuid). - Looks up in lru_cache, then (redis, memcached), then mysql. + Looks up in the 30-second TTL cache, then (redis, memcached), then mysql. Raises UserDoesntExistError if user is not found. (the * makes all arguments keyword-only arguments) """ @@ -316,8 +323,7 @@ class UserManager: # If we change something about a user, we should update the in-memory caches self.set_user_inmemory_cache(user) - # There is no way to clear a single key from the lru_cache... - # https://bugs.python.org/issue28178 + # Clear local cached copies so the blocked state is returned immediately. self.cache_clear() return True |