import redis from redis.lock import Lock from apple_charts_notifications.constants import APP_NAME, REDIS_KEY_TTL, REDIS_LOCK_TTL def get_key(key: str) -> str: """Compose a Redis key. Args: key (str): Key. Returns: str: Complete key constructed by concatenated prefix + key. """ return f'{APP_NAME}/{key}' def get_redis_lock(redis_client: redis.StrictRedis) -> Lock: """Get redis lock""" return redis_client.lock(get_key('lock'), timeout=REDIS_LOCK_TTL) def is_new_track( date: str, label: str, market: str, user_id: str, song_id: str, redis_client: redis.StrictRedis ) -> bool: """Checking is track new or we have already fetched it today Args: date (str): Date with %Y-%m-%d format. label (str): One of added, dropped, moves. market (str): Market short name. user_id (str): User ID. song_id (str): Song ID. redis_client (redis.StrictRedis): Client for Redis. Returns: Boolean """ redis_key = get_key(f'{date}/{market}/{user_id}/{label}') if redis_client.sismember(redis_key, song_id): return False redis_client.sadd(redis_key, song_id) redis_client.expire(redis_key, REDIS_KEY_TTL) return True