github_archive_11.py 1.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. def archive_repo(self, repo: Repository.Repository, repo_path: str, operation: str) -> Optional[str]:
  2. """Clone and pull repos based on the operation passed.
  3. We return the name of the repo if its git operation fails, otherwise return None.
  4. """
  5. logger = woodchips.get(LOGGER_NAME)
  6. failed_repo = None
  7. full_repo_name = os.path.join(repo.owner.login, repo.name) # We use a path here to properly remove failed dirs
  8. if (os.path.exists(repo_path) and operation == CLONE_OPERATION) or (
  9. not os.path.exists(repo_path) and operation == PULL_OPERATION
  10. ):
  11. pass
  12. else:
  13. commands = {
  14. PULL_OPERATION: ['git', '-C', repo_path, 'pull', '--rebase'],
  15. }
  16. if self.use_https:
  17. commands.update({CLONE_OPERATION: ['git', 'clone', repo.html_url, repo_path]})
  18. else:
  19. commands.update({CLONE_OPERATION: ['git', 'clone', repo.ssh_url, repo_path]})
  20. git_command = commands[operation]
  21. try:
  22. subprocess.run(
  23. git_command,
  24. stdout=subprocess.DEVNULL,
  25. stdin=subprocess.DEVNULL,
  26. stderr=subprocess.DEVNULL,
  27. check=True,
  28. timeout=self.timeout,
  29. )
  30. logger.info(f'Repo: {full_repo_name} {operation} success!')
  31. except subprocess.TimeoutExpired:
  32. logger.error(f'Git operation timed out archiving {repo.name}.')
  33. failed_repo = full_repo_name
  34. except subprocess.CalledProcessError as error:
  35. logger.error(f'Failed to {operation} {repo.name}\n{error}')
  36. failed_repo = full_repo_name
  37. return failed_repo