Coverage for langsmith/_internal/_cache_utils.py: 0%

1"""Provide utilities for managing caching in LangSmith. 

2 

3Includes a CacheManager class that handles the lifecycle of caching 

4operations, allowing for easy setup and teardown of caching contexts. 

5""" 

6 

7import pathlib 

8from typing import Optional, Sequence, Union 

9 

10from langsmith import utils 

11 

12 

13class CacheManager: 

14 """Manage caching operations for LangSmith. 

15 

16 Provides methods to start and stop caching, and can be used 

17 as a context manager for automatic cache management. 

18 

19 Attributes: 

20 path (Optional[Union[str, pathlib.Path]]): The path to the cache file. 

21 ignore_hosts (Optional[Sequence[str]]): A list of hosts to ignore in caching. 

22 context_manager: The context manager for caching operations. 

23 context: The context object for the current caching session. 

24 """ 

25 

26 def __init__( 

27 self, 

28 path: Optional[Union[str, pathlib.Path]], 

29 ignore_hosts: Optional[Sequence[str]] = None, 

30 ): 

31 """Initialize the CacheManager. 

32 

33 Args: 

34 path: The path to the cache file. 

35 ignore_hosts: A list of hosts to ignore in caching. 

36 """ 

37 self.path = path 

38 self.ignore_hosts = ignore_hosts 

39 self.context_manager = None 

40 self.context = None 

41 

42 def start_caching(self): 

43 """Start the caching session. 

44 

45 Returns: 

46 self: The CacheManager instance. 

47 

48 Raises: 

49 RuntimeError: If caching is already started. 

50 """ 

51 if self.context is not None: 

52 raise RuntimeError("Caching is already started") 

53 

54 self.context_manager = utils.with_optional_cache(self.path, self.ignore_hosts) 

55 if self.context_manager: 

56 self.context = self.context_manager.__enter__() 

57 return self 

58 

59 def close(self): 

60 """Close the current caching session. 

61 

62 Raises: 

63 RuntimeError: If caching is not started. 

64 """ 

65 if self.context_manager is not None: 

66 self.context_manager.__exit__(None, None, None) 

67 if self.context is None: 

68 raise RuntimeError("Caching is not started") 

69 self.context = None 

70 self.context_manager = None 

71 

72 def __enter__(self): 

73 """Enter the context manager, starting the caching session. 

74 

75 Returns: 

76 self: The CacheManager instance with caching started. 

77 """ 

78 return self.start_caching() 

79 

80 def __exit__(self, exc_type, exc_value, traceback): 

81 """Exit the context manager, closing the caching session. 

82 

83 Args: 

84 exc_type: The type of the exception that caused the context to be exited. 

85 exc_value: The instance of the exception. 

86 traceback: A traceback object encoding the stack trace. 

87 """ 

88 self.close()