Coverage for src / local_deep_research / database / models / cache.py: 91%

1""" 

2Cache model for storing expensive operation results. 

3""" 

4 

5from datetime import datetime, timedelta, UTC 

6from functools import partial 

7 

8from sqlalchemy import JSON, Column, Index, Integer, String, Text 

9from sqlalchemy_utc import UtcDateTime 

10 

11from .base import Base 

12 

13 

14class Cache(Base): 

15 """ 

16 Cache for API responses and expensive operations. 

17 Helps reduce API calls and improve performance. 

18 """ 

19 

20 __tablename__ = "cache" 

21 

22 id = Column(Integer, primary_key=True) 

23 cache_key = Column(String(255), unique=True, nullable=False, index=True) 

24 cache_value = Column(JSON) # For structured data 

25 cache_text = Column(Text) # For large text content 

26 

27 # Cache metadata 

28 cache_type = Column( 

29 String(50) 

30 ) # api_response, computation, search_result, etc. 

31 source = Column(String(100)) # openai, google, computation, etc. 

32 size_bytes = Column(Integer) # Size of cached data 

33 

34 # Expiration 

35 ttl_seconds = Column(Integer) # Time to live in seconds 

36 expires_at = Column(UtcDateTime, index=True) 

37 

38 # Usage tracking 

39 hit_count = Column(Integer, default=0) 

40 created_at = Column(UtcDateTime, default=partial(datetime.now, UTC)) 

41 accessed_at = Column(UtcDateTime, default=partial(datetime.now, UTC)) 

42 

43 # Indexes for performance 

44 __table_args__ = ( 

45 Index("idx_type_expires", "cache_type", "expires_at"), 

46 Index("idx_source_key", "source", "cache_key"), 

47 ) 

48 

49 def is_expired(self) -> bool: 

50 """Check if cache entry has expired.""" 

51 if not self.expires_at: 

52 return False 

53 

54 # Handle both timezone-aware and naive datetimes 

55 now = datetime.now(UTC) 

56 expires = self.expires_at 

57 

58 # If expires_at is naive, make it aware (assuming UTC) 

59 if expires.tzinfo is None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 expires = expires.replace(tzinfo=UTC) 

61 

62 return now > expires 

63 

64 def set_ttl(self, seconds: int): 

65 """Set time to live for cache entry.""" 

66 self.ttl_seconds = seconds 

67 self.expires_at = datetime.now(UTC) + timedelta(seconds=seconds) 

68 

69 def record_hit(self): 

70 """Record a cache hit.""" 

71 self.hit_count += 1 

72 self.accessed_at = datetime.now(UTC) 

73 

74 def __repr__(self): 

75 expired = " (expired)" if self.is_expired() else "" 

76 return f"<Cache(key='{self.cache_key}', type='{self.cache_type}', hits={self.hit_count}{expired})>" 

77 

78 

79class SearchCache(Base): 

80 """Search cache for storing query results with TTL and LRU eviction.""" 

81 

82 __tablename__ = "search_cache" 

83 

84 query_hash = Column(String, primary_key=True) 

85 query_text = Column(Text, nullable=False) 

86 results = Column( 

87 JSON, nullable=False 

88 ) # JSON column for automatic serialization 

89 created_at = Column(Integer, nullable=False) # Unix timestamp 

90 expires_at = Column(Integer, nullable=False) # Unix timestamp 

91 access_count = Column(Integer, default=1) 

92 last_accessed = Column(Integer, nullable=False) # Unix timestamp 

93 

94 __table_args__ = ( 

95 Index("idx_expires_at", "expires_at"), 

96 Index("idx_last_accessed", "last_accessed"), 

97 ) 

98 

99 def __repr__(self): 

100 return f"<SearchCache(query_text='{self.query_text[:50]}...', expires_at={self.expires_at})>"