MCPcopy
hub / github.com/langroid/langroid / load_tree_from_github

Method load_tree_from_github

langroid/parsing/repo_loader.py:304–375  ·  view source on GitHub ↗

Get a nested dictionary of GitHub repository file and directory names up to a certain depth, with file contents. Args: depth (int): The depth level. lines (int): The number of lines of file contents to include. Returns: Dict[str,

(
        self, depth: int, lines: int = 0
    )

Source from the content-addressed store, hash-verified

302 return self.clone_path
303
304 def load_tree_from_github(
305 self, depth: int, lines: int = 0
306 ) -> Dict[str, Union[str, List[Dict[str, Any]]]]:
307 """
308 Get a nested dictionary of GitHub repository file and directory names
309 up to a certain depth, with file contents.
310
311 Args:
312 depth (int): The depth level.
313 lines (int): The number of lines of file contents to include.
314
315 Returns:
316 Dict[str, Union[str, List[Dict]]]:
317 A dictionary containing file and directory names, with file contents.
318 """
319 if self.repo is None:
320 logger.warning("No repo found. Ensure the URL is correct.")
321 return {} # Return an empty dict rather than raise an error in this case
322
323 root_contents = self.repo.get_contents("")
324 if not isinstance(root_contents, list):
325 root_contents = [root_contents]
326 repo_structure = {
327 "type": "dir",
328 "name": "",
329 "dirs": [],
330 "files": [],
331 "path": "",
332 }
333
334 # A queue of tuples (current_node, current_depth, parent_structure)
335 queue = deque([(root_contents, 0, repo_structure)])
336
337 while queue:
338 current_node, current_depth, parent_structure = queue.popleft()
339
340 for content in current_node:
341 if not self._is_allowed(content):
342 continue
343 if content.type == "dir" and current_depth < depth:
344 # Create a new sub-dictionary for this directory
345 new_dir = {
346 "type": "dir",
347 "name": content.name,
348 "dirs": [],
349 "files": [],
350 "path": content.path,
351 }
352 parent_structure["dirs"].append(new_dir)
353 contents = self.repo.get_contents(content.path)
354 if not isinstance(contents, list):
355 contents = [contents]
356 queue.append(
357 (
358 contents,
359 current_depth + 1,
360 new_dir,
361 )

Callers 1

test_repo_loaderFunction · 0.95

Calls 2

_is_allowedMethod · 0.95
_get_decoded_contentFunction · 0.85

Tested by 1

test_repo_loaderFunction · 0.76