Coverage for app / security / user_session_cache.py: 67%

30 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2025-12-06 04:49 +0000

1""" 

2This module provides utilities to manage user sessions and caching in a Flask application. 

3 

4It includes methods for loading users from caching or the datastore, handling logout events, 

5and generating cache keys. These tools improve performance while supporting Flask-Security  

6integration. 

7""" 

8from flask import session 

9from flask_login import AnonymousUserMixin 

10from flask_security.utils import set_request_attr 

11 

12 

13# Custom user loader that utilizes cache 

14def custom_user_loader(user_id): 

15 """Load user, using SimpleCache to store and retrieve the user object.""" 

16 # pylint: disable=import-outside-toplevel 

17 from app import cache # noqa: E402 

18 

19 # Attempt to retrieve the user from the cache 

20 user = cache.get(_cache_key_from_string(user_id)) 

21 if user is None: 

22 # If not in cache, retrieve from datastore and cache it 

23 user = _load_user_from_datastore(user_id) 

24 if user: 24 ↛ 26line 24 didn't jump to line 26 because the condition on line 24 was always true

25 cache.set(_cache_key_from_string(user_id), user) 

26 if user: 26 ↛ 30line 26 didn't jump to line 30 because the condition on line 26 was always true

27 # Set Flask-Security session-related attributes (if any) 

28 set_request_attr("fs_authn_via", "session") 

29 set_request_attr("fs_paa", session.get("fs_paa", 0)) 

30 return user 

31 

32 

33# noinspection PyUnusedLocal, PyUnusedParameter 

34def on_logout(_sender, user): # noqa 

35 """Callback for when a user logs out.""" 

36 # pylint: disable=import-outside-toplevel 

37 from app import cache 

38 

39 # '_sender' is unused but required by the signal callback API 

40 if user: 

41 # Invalidate the cache for the user 

42 cache.delete(_cache_key_from_user(user)) 

43 

44 

45def _cache_key_from_string(fs_uniquifier: str): 

46 return f"user_{fs_uniquifier}" 

47 

48 

49def _cache_key_from_user(user): 

50 if isinstance(user, AnonymousUserMixin): 

51 # Handle the case where the user is not authenticated 

52 return None 

53 return _cache_key_from_string(user.fs_uniquifier) 

54 

55 

56def _load_user_from_datastore(user_id): 

57 """ 

58 Loads a user from the datastore if not already cached. 

59 """ 

60 # pylint: disable=import-outside-toplevel 

61 from app import user_datastore 

62 

63 # Assumes fs_uniquifier is used for user lookup 

64 user = user_datastore.find_user(fs_uniquifier=str(user_id)) 

65 if user and user.active: 65 ↛ 67line 65 didn't jump to line 67 because the condition on line 65 was always true

66 return user 

67 return None