|
@@ -1,27 +1,71 @@
|
|
|
-from typing import Union, Iterable
|
|
|
+import asyncio
|
|
|
from shlex import quote
|
|
|
-from pathlib import Path
|
|
|
+from subprocess import SubprocessError
|
|
|
+from dataclasses import dataclass
|
|
|
from asyncio.subprocess import Process
|
|
|
+from typing import Iterable, Any, cast
|
|
|
|
|
|
from .shell import Shell
|
|
|
|
|
|
|
|
|
-class Repo(Shell):
|
|
|
- pwd: Path
|
|
|
+class GitShell(Shell):
|
|
|
+ async def cmd(self, args, **kwargs) -> Process:
|
|
|
+ params = self._kwargs_to_params(**kwargs)
|
|
|
+
|
|
|
+ return await self.run(('git', *args, *params))
|
|
|
+
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class ConfigSection(GitShell):
|
|
|
+ section: str
|
|
|
+
|
|
|
+ async def value(self, name: str) -> str:
|
|
|
+ try:
|
|
|
+ process = await self.cmd(('config', '--get', f'{self.section}.{name}'))
|
|
|
+ except SubprocessError:
|
|
|
+ return ''
|
|
|
|
|
|
- def __init__(self, path: Union[str, Path]):
|
|
|
- super().__init__()
|
|
|
+ return cast(str, process.stdout)
|
|
|
|
|
|
- self.cd(path)
|
|
|
+ async def set(self, name: str, value: Any) -> Any:
|
|
|
+ await self.cmd(('config', f'{self.section}.{name}', quote(value)))
|
|
|
|
|
|
+ return self
|
|
|
+
|
|
|
+
|
|
|
+class Config(GitShell):
|
|
|
+ def section(self, name: str) -> ConfigSection:
|
|
|
+ return ConfigSection(self.pwd, name)
|
|
|
+
|
|
|
+
|
|
|
+class Repo(GitShell):
|
|
|
@property
|
|
|
def initialized(self) -> bool:
|
|
|
return (self.pwd / '.git').exists()
|
|
|
|
|
|
- async def cmd(self, args, **kwargs) -> Process:
|
|
|
- params = self._kwargs_to_params(**kwargs)
|
|
|
+ @property
|
|
|
+ def config(self) -> Config:
|
|
|
+ return Config(self.pwd)
|
|
|
+
|
|
|
+ async def setup(self, email: str = '', name: str = ''):
|
|
|
+ tasks = []
|
|
|
+
|
|
|
+ section = self.config.section('user')
|
|
|
+
|
|
|
+ cur_name = await section.value('name')
|
|
|
+
|
|
|
+ if not cur_name:
|
|
|
+ tasks.append(section.set('name', name))
|
|
|
+
|
|
|
+ cur_email = await section.value('email')
|
|
|
+
|
|
|
+ if not cur_email:
|
|
|
+ tasks.append(section.set('email', email))
|
|
|
+
|
|
|
+ await asyncio.gather(*tasks)
|
|
|
+
|
|
|
+ return self
|
|
|
|
|
|
- return await self.run(('git', *args, *params))
|
|
|
|
|
|
async def init(self) -> Process:
|
|
|
if self.initialized:
|