Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/libcst/metadata/full_repo_manager.py: 37%

1# Copyright (c) Meta Platforms, Inc. and affiliates. 

2# 

3# This source code is licensed under the MIT license found in the 

4# LICENSE file in the root directory of this source tree. 

5 

6 

7from pathlib import Path 

8from typing import Collection, Dict, List, Mapping, TYPE_CHECKING 

9 

10import libcst as cst 

11from libcst._types import StrPath 

12from libcst.metadata.wrapper import MetadataWrapper 

13 

14if TYPE_CHECKING: 

15 from libcst.metadata.base_provider import ProviderT # noqa: F401 

16 

17 

18class FullRepoManager: 

19 def __init__( 

20 self, 

21 repo_root_dir: StrPath, 

22 paths: Collection[str], 

23 providers: Collection["ProviderT"], 

24 timeout: int = 5, 

25 ) -> None: 

26 """ 

27 Given project root directory with pyre and watchman setup, :class:`~libcst.metadata.FullRepoManager` 

28 handles the inter process communication to read the required full repository cache data for 

29 metadata provider like :class:`~libcst.metadata.TypeInferenceProvider`. 

30 

31 :param paths: a collection of paths to access full repository data. 

32 :param providers: a collection of metadata provider classes require accessing full repository data, currently supports 

33 :class:`~libcst.metadata.TypeInferenceProvider` and 

34 :class:`~libcst.metadata.FullyQualifiedNameProvider`. 

35 :param timeout: number of seconds. Raises `TimeoutExpired <https://docs.python.org/3/library/subprocess.html#subprocess.TimeoutExpired>`_ 

36 when timeout. 

37 """ 

38 self.root_path: Path = Path(repo_root_dir) 

39 self._cache: Dict["ProviderT", Mapping[str, object]] = {} 

40 self._timeout = timeout 

41 self._providers = providers 

42 self._paths: List[str] = list(paths) 

43 

44 @property 

45 def cache(self) -> Dict["ProviderT", Mapping[str, object]]: 

46 """ 

47 The full repository cache data for all metadata providers passed in the ``providers`` parameter when 

48 constructing :class:`~libcst.metadata.FullRepoManager`. Each provider is mapped to a mapping of path to cache. 

49 """ 

50 # Make sure that the cache is available to us. If resolve_cache() was called manually then this is a noop. 

51 self.resolve_cache() 

52 return self._cache 

53 

54 def resolve_cache(self) -> None: 

55 """ 

56 Resolve cache for all providers that require it. Normally this is called by 

57 :meth:`~FullRepoManager.get_cache_for_path` so you do not need to call it 

58 manually. However, if you intend to do a single cache resolution pass before 

59 forking, it is a good idea to call this explicitly to control when cache 

60 resolution happens. 

61 """ 

62 if not self._cache: 

63 cache: Dict["ProviderT", Mapping[str, object]] = {} 

64 for provider in self._providers: 

65 handler = provider.gen_cache 

66 if handler: 

67 cache[provider] = handler( 

68 self.root_path, self._paths, self._timeout 

69 ) 

70 self._cache = cache 

71 

72 def get_cache_for_path(self, path: str) -> Mapping["ProviderT", object]: 

73 """ 

74 Retrieve cache for a source file. The file needs to appear in the ``paths`` parameter when 

75 constructing :class:`~libcst.metadata.FullRepoManager`. 

76 

77 .. code-block:: python 

78 

79 manager = FullRepoManager(".", {"a.py", "b.py"}, {TypeInferenceProvider}) 

80 MetadataWrapper(module, cache=manager.get_cache_for_path("a.py")) 

81 """ 

82 if path not in self._paths: 

83 raise Exception( 

84 "The path needs to be in paths parameter when constructing FullRepoManager for efficient batch processing." 

85 ) 

86 # Make sure that the cache is available to us. If the user called 

87 # resolve_cache() manually then this is a noop. 

88 self.resolve_cache() 

89 return { 

90 provider: data 

91 for provider, files in self._cache.items() 

92 for _path, data in files.items() 

93 if _path == path 

94 } 

95 

96 def get_metadata_wrapper_for_path(self, path: str) -> MetadataWrapper: 

97 """ 

98 Create a :class:`~libcst.metadata.MetadataWrapper` given a source file path. 

99 The path needs to be a path relative to project root directory. 

100 The source code is read and parsed as :class:`~libcst.Module` for 

101 :class:`~libcst.metadata.MetadataWrapper`. 

102 

103 .. code-block:: python 

104 

105 manager = FullRepoManager(".", {"a.py", "b.py"}, {TypeInferenceProvider}) 

106 wrapper = manager.get_metadata_wrapper_for_path("a.py") 

107 """ 

108 module = cst.parse_module((self.root_path / path).read_text()) 

109 cache = self.get_cache_for_path(path) 

110 return MetadataWrapper(module, True, cache)