Coverage for nexios\session\file.py: 64%

1import os 

2import json 

3from datetime import datetime, timezone 

4from typing import Optional, Dict, Any 

5from .base import BaseSessionInterface 

6 

7 

8class FileSessionManager(BaseSessionInterface): 

9 def __init__(self, session_key: str) -> None: 

10 super().__init__(session_key) 

11 self.session_key = session_key or self.get_session_key() 

12 self.session_file_path = os.path.join( 

13 self.session_config.session_file_name or "sessions", 

14 f"{self.session_key}.json", 

15 ) 

16 

17 # Ensure the session storage directory exists 

18 os.makedirs( 

19 self.session_config.session_file_storage_path or "sessions", exist_ok=True 

20 ) 

21 

22 def _load_session_data(self) -> Optional[Dict[str, Any]]: 

23 """Load session data from the file.""" 

24 if os.path.exists(self.session_file_path): 

25 with open(self.session_file_path, "r") as file: 

26 try: 

27 session_data = json.load(file) 

28 return session_data 

29 except json.JSONDecodeError: 

30 return None 

31 return None 

32 

33 def _save_session_data(self): 

34 """Save the session data to a file.""" 

35 with open(self.session_file_path, "w") as file: 

36 json.dump(self._session_cache, file) 

37 

38 def set_session(self, key: str, value: str): 

39 """Set a session value.""" 

40 self.modified = True 

41 self._session_cache[key] = value 

42 

43 def get_session(self, key: str) -> Optional[str]: 

44 """Get a session value.""" 

45 return self._session_cache.get(key, None) 

46 

47 def get_all(self): 

48 """Get all session data.""" 

49 return self._session_cache.items() 

50 

51 def keys(self): 

52 """Get all session keys.""" 

53 return self._session_cache.keys() 

54 

55 def values(self): 

56 """Get all session values.""" 

57 return self._session_cache.values() 

58 

59 def is_empty(self) -> bool: 

60 """Check if the session is empty.""" 

61 return len(self._session_cache.items()) == 0 

62 

63 async def save(self): # type: ignore 

64 """Save the session data to the file.""" 

65 self._save_session_data() 

66 

67 @property 

68 def should_set_cookie(self) -> bool: 

69 """Determines if the cookie should be set.""" 

70 return self.modified or ( 

71 self.session_config.session_permanent 

72 and self.session_config.session_refresh_each_request 

73 ) 

74 

75 def has_expired(self) -> bool: 

76 """Returns True if the session has expired.""" 

77 expiration_time = self.get_expiration_time() 

78 if ( 

79 expiration_time and datetime.now(timezone.utc) > expiration_time 

80 ): # type:ignore 

81 return True 

82 return False 

83 

84 async def load(self): 

85 """Load the session data from the file.""" 

86 session_data = self._load_session_data() 

87 

88 if session_data: 

89 self._session_cache.update(session_data) 

90 else: 

91 self._session_cache = {} 

92 

93 def clear(self): 

94 """Clear the session data.""" 

95 self._session_cache.clear() 

96 if os.path.exists(self.session_file_path): 

97 os.remove(self.session_file_path)