git.py 1.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. from typing import Union, Iterable
  2. from shlex import quote
  3. from pathlib import Path
  4. from asyncio.subprocess import Process
  5. from .shell import Shell
  6. class Repo(Shell):
  7. pwd: Path
  8. def __init__(self, path: Union[str, Path]):
  9. super().__init__()
  10. self.cd(path)
  11. @property
  12. def initialized(self) -> bool:
  13. return (self.pwd / '.git').exists()
  14. async def cmd(self, args, **kwargs) -> Process:
  15. params = self._kwargs_to_params(**kwargs)
  16. return await self.run(('git', *args, *params))
  17. async def init(self) -> Process:
  18. if self.initialized:
  19. raise ValueError(
  20. f'There is already a git repository at {self.pwd}'
  21. )
  22. return await self.cmd(('init',))
  23. async def add(self, files: Iterable[str], **kwargs) -> Process:
  24. args = ('add', *map(quote, files))
  25. return await self.cmd(args, **kwargs)
  26. async def commit(self, message: str, **kwargs) -> Process:
  27. args = ('commit', '-m', quote(message))
  28. return await self.cmd(args, **kwargs)