Coverage for session_buddy / backends / base.py: 85.37%

1"""Base classes for serverless session storage backends. 

2 

3This module provides the abstract base class and data models for session storage 

4backends including Redis, S3, local file storage, and ACB cache. 

5""" 

6 

7from __future__ import annotations 

8 

9import gzip 

10import logging 

11from abc import ABC, abstractmethod 

12from datetime import datetime 

13from typing import Any 

14 

15from pydantic import BaseModel, Field, field_validator 

16 

17 

18class SessionState(BaseModel): 

19 """Represents complete session state for serialization.""" 

20 

21 session_id: str = Field( 

22 min_length=1, 

23 description="Unique identifier for the session", 

24 ) 

25 user_id: str = Field(min_length=1, description="Identifier for the user") 

26 project_id: str = Field(min_length=1, description="Identifier for the project") 

27 created_at: str = Field(description="ISO timestamp when session was created") 

28 last_activity: str = Field(description="ISO timestamp of last activity") 

29 permissions: list[str] = Field( 

30 default_factory=list, 

31 description="List of permissions granted to the session", 

32 ) 

33 conversation_history: list[dict[str, Any]] = Field( 

34 default_factory=list, 

35 description="History of conversation entries", 

36 ) 

37 reflection_data: dict[str, Any] = Field( 

38 default_factory=dict, 

39 description="Stored reflection and memory data", 

40 ) 

41 app_monitoring_state: dict[str, Any] = Field( 

42 default_factory=dict, 

43 description="Application monitoring state", 

44 ) 

45 llm_provider_configs: dict[str, Any] = Field( 

46 default_factory=dict, 

47 description="LLM provider configurations", 

48 ) 

49 metadata: dict[str, Any] = Field( 

50 default_factory=dict, 

51 description="Additional session metadata", 

52 ) 

53 

54 @field_validator("created_at", "last_activity") 

55 @classmethod 

56 def validate_iso_timestamp(cls, v: str) -> str: 

57 """Validate that timestamps are in ISO format.""" 

58 try: 

59 datetime.fromisoformat(v) 

60 return v 

61 except ValueError as e: 

62 msg = f"Invalid ISO timestamp format: {v}" 

63 raise ValueError(msg) from e 

64 

65 def to_dict(self) -> dict[str, Any]: 

66 """Convert to dictionary for serialization.""" 

67 return self.model_dump() 

68 

69 @classmethod 

70 def from_dict(cls, data: dict[str, Any]) -> SessionState: 

71 """Create from dictionary.""" 

72 return cls.model_validate(data) 

73 

74 def get_compressed_size(self) -> int: 

75 """Get compressed size of session state.""" 

76 serialized = self.model_dump_json() 

77 compressed = gzip.compress(serialized.encode("utf-8")) 

78 return len(compressed) 

79 

80 

81class SessionStorage(ABC): 

82 """Abstract base class for session storage backends.""" 

83 

84 def __init__(self, config: dict[str, Any]) -> None: 

85 self.config = config 

86 self.logger = logging.getLogger(f"serverless.{self.__class__.__name__.lower()}") 

87 

88 @abstractmethod 

89 async def store_session( 

90 self, 

91 session_state: SessionState, 

92 ttl_seconds: int | None = None, 

93 ) -> bool: 

94 """Store session state with optional TTL.""" 

95 

96 @abstractmethod 

97 async def retrieve_session(self, session_id: str) -> SessionState | None: 

98 """Retrieve session state by ID.""" 

99 

100 @abstractmethod 

101 async def delete_session(self, session_id: str) -> bool: 

102 """Delete session state.""" 

103 

104 @abstractmethod 

105 async def list_sessions( 

106 self, 

107 user_id: str | None = None, 

108 project_id: str | None = None, 

109 ) -> list[str]: 

110 """List session IDs matching criteria.""" 

111 

112 @abstractmethod 

113 async def cleanup_expired_sessions(self) -> int: 

114 """Clean up expired sessions, return count removed.""" 

115 

116 @abstractmethod 

117 async def is_available(self) -> bool: 

118 """Check if storage backend is available."""