github_archive_12.py 1.6 KB

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