1234567891011121314151617181920212223242526272829303132333435363738 |
- def archive_gist(self, gist: Gist.Gist, gist_path: str, operation: str) -> Optional[str]:
- """Clone and pull gists based on the operation passed.
- We return the name of the gist if its git operation fails, otherwise return None.
- """
- logger = woodchips.get(LOGGER_NAME)
- failed_gist = None
- full_gist_id = os.path.join(gist.owner.login, gist.id) # We use a path here to properly remove failed dirs
- if (os.path.exists(gist_path) and operation == CLONE_OPERATION) or (
- not os.path.exists(gist_path) and operation == PULL_OPERATION
- ):
- pass
- else:
- commands = {
- CLONE_OPERATION: ['git', 'clone', gist.html_url, gist_path],
- PULL_OPERATION: ['git', '-C', gist_path, 'pull', '--rebase'],
- }
- git_command = commands[operation]
- try:
- subprocess.run(
- git_command,
- stdout=subprocess.DEVNULL,
- stdin=subprocess.DEVNULL,
- stderr=subprocess.DEVNULL,
- check=True,
- timeout=self.timeout,
- )
- logger.info(f'Gist: {full_gist_id} {operation} success!')
- except subprocess.TimeoutExpired:
- logger.error(f'Git operation timed out archiving {gist.id}.')
- failed_gist = full_gist_id
- except subprocess.CalledProcessError as error:
- logger.error(f'Failed to {operation} {gist.id}\n{error}')
- failed_gist = full_gist_id
- return failed_gist
|