24346692409

Committed 13 Apr 2026 01:43PM UTC coverage: 80.417% (+0.2%) from 80.257%

Build # 24346692409

Build Type

push

github

Committed by

web-flow

Commit Message

Version 0.7.1 - Bug fixes and UX improvements (#30)

* chore: update release notes for version 0.7.1

* feat: enhance collection handling across UI components to prevent actions without a selected collection

* test: add tests for collection ready state and telemetry event handling

* docs: update release notes for version 0.7.1 to reflect bug fixes and UX improvements

* refactor: streamline profile connection handling and remove unused dialog prompt

* chore: update version to 0.7.1 in pyproject.toml and __init__.py

* refactor: improve text file detection logic in is_text_file function

* test: add tests for connection handling and collection state updates

* refactor: enhance tooltips for search and visualization buttons, improve text file detection logic

Coverage Stats

63 of 68 new or added lines in 8 files covered. (92.65%)

5 existing lines in 1 file now uncovered.

14508 of 18041 relevant lines covered (80.42%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.67

/src/vector_inspector/core/cache_manager.py

"""
Cache manager for storing databrowser and search panel state by database and collection.
Provides fast switching between collections with automatic invalidation on refresh or settings changes.
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional


@dataclass
class CacheEntry:
    """Represents a cached state for a specific database and collection."""

    data: Any
    timestamp: datetime = field(default_factory=datetime.now)

    # Browser state
    scroll_position: int = 0
    selected_indices: list = field(default_factory=list)

    # Search panel state
    search_query: str = ""
    search_filters: dict[str, Any] = field(default_factory=dict)
    search_results: Optional[Any] = None

    # User inputs
    user_inputs: dict[str, Any] = field(default_factory=dict)


class CacheManager:
    """
    Manages cache for databrowser and search panel by (database, collection) key.
    Supports invalidation on refresh or settings changes.

    Uses singleton pattern to ensure only one instance exists, preventing state inconsistencies
    when code creates instances directly vs using AppState.
    """

    _instance: Optional["CacheManager"] = None

    def __new__(cls):
        """Ensure only one instance exists (singleton pattern)."""
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        """Initialize cache manager (only runs once due to singleton)."""
        # Skip if already initialized (check instance dict only, not class dict)
        if "_initialized" in self.__dict__:
            return
        self._initialized = True

        self._cache: dict[tuple[str, str], CacheEntry] = {}
        self._enabled = True

    def get(self, database: str, collection: str) -> Optional[CacheEntry]:
        """Retrieve cached entry for a database and collection."""
        if not self._enabled:
            return None

        key = (database, collection)
        return self._cache.get(key)

    def set(self, database: str, collection: str, entry: CacheEntry) -> None:
        """Store a cache entry for a database and collection."""
        if not self._enabled:
            return

        key = (database, collection)
        entry.timestamp = datetime.now()
        self._cache[key] = entry

    def update(self, database: str, collection: str, **kwargs) -> None:
        """Update specific fields in an existing cache entry."""
        key = (database, collection)
        if key in self._cache:
            entry = self._cache[key]
            for field_name, value in kwargs.items():
                if hasattr(entry, field_name):
                    setattr(entry, field_name, value)
            entry.timestamp = datetime.now()
        else:
            # Create new entry with provided fields
            entry = CacheEntry(data=None)
            for field_name, value in kwargs.items():
                if hasattr(entry, field_name):
                    setattr(entry, field_name, value)
            self._cache[key] = entry

    def invalidate(self, database: Optional[str] = None, collection: Optional[str] = None) -> None:
        """
        Invalidate cache entries.
        - If both database and collection are provided, invalidate that specific entry.
        - If only database is provided, invalidate all collections in that database.
        - If neither is provided, invalidate all entries (global refresh).
        """
        if database is None and collection is None:
            # Clear all cache
            self._cache.clear()
        elif collection is None and database is not None:
            # Clear all collections in the specified database
            keys_to_remove = [key for key in self._cache.keys() if key[0] == database]
            for key in keys_to_remove:
                del self._cache[key]
        elif database is not None and collection is not None:
            # Clear specific database/collection combination
            key = (database, collection)
            if key in self._cache:
                del self._cache[key]

    def clear(self) -> None:
        """Clear all cached entries."""
        self._cache.clear()

    def enable(self) -> None:
        """Enable caching."""
        self._enabled = True

    def disable(self) -> None:
        """Disable caching and clear all entries."""
        self._enabled = False
        self._cache.clear()

    def is_enabled(self) -> bool:
        """Check if caching is enabled."""
        return self._enabled

    def get_cache_info(self) -> dict[str, Any]:
        """Get information about the current cache state."""
        return {
            "enabled": self._enabled,
            "entry_count": len(self._cache),
            "entries": [
                {
                    "database": db,
                    "collection": coll,
                    "timestamp": entry.timestamp.isoformat(),
                    "has_data": entry.data is not None,
                    "has_search_results": entry.search_results is not None,
                }
                for (db, coll), entry in self._cache.items()
            ],
        }


# Legacy global function for backward compatibility
# DEPRECATED: New code should use app_state.cache_manager instead
def get_cache_manager() -> CacheManager:
    """Get the singleton cache manager instance.

    DEPRECATED: This returns the singleton instance for legacy code.
    New code should use app_state.cache_manager instead.

    Note: CacheManager uses singleton pattern, so this always returns the same instance
    that AppState uses.
    """
    return CacheManager()


def invalidate_cache_on_settings_change() -> None:
    """Invalidate all cache when settings change."""
    cache = get_cache_manager()
    cache.invalidate()


def invalidate_cache_on_refresh(database: Optional[str] = None, collection: Optional[str] = None) -> None:
    """Invalidate cache on refresh action."""
    cache = get_cache_manager()
    cache.invalidate(database, collection)

1	"""
2	Cache manager for storing databrowser and search panel state by database and collection.
3	Provides fast switching between collections with automatic invalidation on refresh or settings changes.
4	"""
5
6	from dataclasses import dataclass, field	1✔
7	from datetime import datetime	1✔
8	from typing import Any, Optional	1✔
9
10
11	@dataclass	1✔
12	class CacheEntry:	1✔
13	"""Represents a cached state for a specific database and collection."""
14
15	data: Any	1✔
16	timestamp: datetime = field(default_factory=datetime.now)	1✔
17
18	# Browser state
19	scroll_position: int = 0	1✔
20	selected_indices: list = field(default_factory=list)	1✔
21
22	# Search panel state
23	search_query: str = ""	1✔
24	search_filters: dict[str, Any] = field(default_factory=dict)	1✔
25	search_results: Optional[Any] = None	1✔
26
27	# User inputs
28	user_inputs: dict[str, Any] = field(default_factory=dict)	1✔
29
30
31	class CacheManager:	1✔
32	"""
33	Manages cache for databrowser and search panel by (database, collection) key.
34	Supports invalidation on refresh or settings changes.
35
36	Uses singleton pattern to ensure only one instance exists, preventing state inconsistencies
37	when code creates instances directly vs using AppState.
38	"""
39
40	_instance: Optional["CacheManager"] = None	1✔
41
42	def __new__(cls):	1✔
43	"""Ensure only one instance exists (singleton pattern)."""
44	if cls._instance is None:	1✔
45	cls._instance = super().__new__(cls)	1✔
46	return cls._instance	1✔
47
48	def __init__(self):	1✔
49	"""Initialize cache manager (only runs once due to singleton)."""
50	# Skip if already initialized (check instance dict only, not class dict)
51	if "_initialized" in self.__dict__:	1✔
52	return	1✔
53	self._initialized = True	1✔
54
55	self._cache: dict[tuple[str, str], CacheEntry] = {}	1✔
56	self._enabled = True	1✔
57
58	def get(self, database: str, collection: str) -> Optional[CacheEntry]:	1✔
59	"""Retrieve cached entry for a database and collection."""
60	if not self._enabled:	1✔
61	return None	1✔
62
63	key = (database, collection)	1✔
64	return self._cache.get(key)	1✔
65
66	def set(self, database: str, collection: str, entry: CacheEntry) -> None:	1✔
67	"""Store a cache entry for a database and collection."""
68	if not self._enabled:	1✔
69	return	1✔
70
71	key = (database, collection)	1✔
72	entry.timestamp = datetime.now()	1✔
73	self._cache[key] = entry	1✔
74
75	def update(self, database: str, collection: str, **kwargs) -> None:	1✔
76	"""Update specific fields in an existing cache entry."""
77	key = (database, collection)	1✔
78	if key in self._cache:	1✔
UNCOV 79	entry = self._cache[key]	×
UNCOV 80	for field_name, value in kwargs.items():	×
UNCOV 81	if hasattr(entry, field_name):	×
UNCOV 82	setattr(entry, field_name, value)	×
UNCOV 83	entry.timestamp = datetime.now()	×
84	else:
85	# Create new entry with provided fields
86	entry = CacheEntry(data=None)	1✔
87	for field_name, value in kwargs.items():	1✔
88	if hasattr(entry, field_name):	1✔
89	setattr(entry, field_name, value)	1✔
90	self._cache[key] = entry	1✔
91
92	def invalidate(self, database: Optional[str] = None, collection: Optional[str] = None) -> None:	1✔
93	"""
94	Invalidate cache entries.
95	- If both database and collection are provided, invalidate that specific entry.
96	- If only database is provided, invalidate all collections in that database.
97	- If neither is provided, invalidate all entries (global refresh).
98	"""
99	if database is None and collection is None:	1✔
100	# Clear all cache
101	self._cache.clear()	1✔
102	elif collection is None and database is not None:	1✔
103	# Clear all collections in the specified database
104	keys_to_remove = [key for key in self._cache.keys() if key[0] == database]	1✔
105	for key in keys_to_remove:	1✔
106	del self._cache[key]	1✔
107	elif database is not None and collection is not None:	1✔
108	# Clear specific database/collection combination
109	key = (database, collection)	1✔
110	if key in self._cache:	1✔
111	del self._cache[key]	1✔
112
113	def clear(self) -> None:	1✔
114	"""Clear all cached entries."""
115	self._cache.clear()	1✔
116
117	def enable(self) -> None:	1✔
118	"""Enable caching."""
119	self._enabled = True	1✔
120
121	def disable(self) -> None:	1✔
122	"""Disable caching and clear all entries."""
123	self._enabled = False	1✔
124	self._cache.clear()	1✔
125
126	def is_enabled(self) -> bool:	1✔
127	"""Check if caching is enabled."""
128	return self._enabled	1✔
129
130	def get_cache_info(self) -> dict[str, Any]:	1✔
131	"""Get information about the current cache state."""
132	return {	1✔
133	"enabled": self._enabled,
134	"entry_count": len(self._cache),
135	"entries": [
136	{
137	"database": db,
138	"collection": coll,
139	"timestamp": entry.timestamp.isoformat(),
140	"has_data": entry.data is not None,
141	"has_search_results": entry.search_results is not None,
142	}
143	for (db, coll), entry in self._cache.items()
144	],
145	}
146
147
148	# Legacy global function for backward compatibility
149	# DEPRECATED: New code should use app_state.cache_manager instead
150	def get_cache_manager() -> CacheManager:	1✔
151	"""Get the singleton cache manager instance.
152
153	DEPRECATED: This returns the singleton instance for legacy code.
154	New code should use app_state.cache_manager instead.
155
156	Note: CacheManager uses singleton pattern, so this always returns the same instance
157	that AppState uses.
158	"""
159	return CacheManager()	1✔
160
161
162	def invalidate_cache_on_settings_change() -> None:	1✔
163	"""Invalidate all cache when settings change."""
164	cache = get_cache_manager()	1✔
165	cache.invalidate()	1✔
166
167
168	def invalidate_cache_on_refresh(database: Optional[str] = None, collection: Optional[str] = None) -> None:	1✔
169	"""Invalidate cache on refresh action."""
170	cache = get_cache_manager()	1✔
171	cache.invalidate(database, collection)	1✔

anthonypdawson / vector-inspector / 24346692409

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous