mirror of
https://github.com/karma-riuk/crab.git
synced 2025-07-05 05:28:13 +02:00
made clone use raising expections instead of updates
This commit is contained in:
53
utils.py
53
utils.py
@ -5,6 +5,8 @@ from github.PaginatedList import PaginatedList
|
|||||||
from github.PullRequestComment import PullRequestComment
|
from github.PullRequestComment import PullRequestComment
|
||||||
from tqdm import tqdm
|
from tqdm import tqdm
|
||||||
|
|
||||||
|
from errors import CantCloneRepoError
|
||||||
|
|
||||||
|
|
||||||
def move_github_logging_to_file():
|
def move_github_logging_to_file():
|
||||||
github_logger = logging.getLogger("github")
|
github_logger = logging.getLogger("github")
|
||||||
@ -128,46 +130,31 @@ def is_already_repo_cloned(repos_dir: str, repo_name: str) -> bool:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
def clone(
|
def clone(repo: str, dest: str, force: bool = False) -> None:
|
||||||
repo: str, dest: str, updates: dict = {}, force: bool = False, verbose: bool = False
|
|
||||||
) -> bool:
|
|
||||||
"""
|
"""
|
||||||
Clones a GitHub repository into a local directory.
|
Clone a GitHub repository to the specified destination directory.
|
||||||
|
|
||||||
Args:
|
Parameters:
|
||||||
repo (str): The GitHub repository to clone, in the format "owner/repo".
|
repo (str): The name of the repository to clone (e.g., "user/repo").
|
||||||
dest (str): The directory to clone the repository into.
|
dest (str): The destination directory where the repository will be cloned.
|
||||||
updates (dict, optional): A dictionary to store updates about the cloning process.
|
force (bool): If True, force clone even if the repository already exists.
|
||||||
force (bool): Whether to force the cloning process, even if the repository already exists.
|
|
||||||
verbose (bool): Whether to print verbose output.
|
|
||||||
|
|
||||||
Returns:
|
Raises:
|
||||||
bool: True if the repository is successfully cloned, False otherwise.
|
CantCloneRepoError: If the repository cannot be cloned.
|
||||||
"""
|
"""
|
||||||
local_repo_path = os.path.join(dest, repo)
|
local_repo_path = os.path.join(dest, repo)
|
||||||
if not force and is_already_repo_cloned(dest, repo):
|
if not force and is_already_repo_cloned(dest, repo):
|
||||||
# if verbose: print(f"Skipping {repo}, already exists")
|
return
|
||||||
updates["cloned_successfully"] = "Already exists"
|
|
||||||
return True
|
|
||||||
|
|
||||||
if verbose:
|
try:
|
||||||
print(f"Cloning {repo}")
|
subprocess.run(
|
||||||
proc = subprocess.run(
|
["git", "clone", "--depth", "1", f"https://github.com/{repo}", local_repo_path],
|
||||||
["git", "clone", "--depth", "1", f"https://github.com/{repo}", local_repo_path],
|
check=True,
|
||||||
stdout=subprocess.PIPE,
|
capture_output=True,
|
||||||
stderr=subprocess.PIPE,
|
text=True,
|
||||||
)
|
)
|
||||||
if proc.returncode != 0:
|
except subprocess.CalledProcessError as e:
|
||||||
updates["cloned_successfully"] = False
|
raise CantCloneRepoError(e.stderr)
|
||||||
print(f"Failed to clone {repo}", file=sys.stderr)
|
|
||||||
print(f"Error message was:", file=sys.stderr)
|
|
||||||
error_msg = proc.stderr.decode()
|
|
||||||
print(error_msg, file=sys.stderr)
|
|
||||||
updates["error_msg"] = error_msg
|
|
||||||
return False
|
|
||||||
else:
|
|
||||||
updates["cloned_successfully"] = True
|
|
||||||
return True
|
|
||||||
|
|
||||||
|
|
||||||
def run_git_cmd(cmd: list[str], repo_path: str) -> subprocess.CompletedProcess:
|
def run_git_cmd(cmd: list[str], repo_path: str) -> subprocess.CompletedProcess:
|
||||||
|
Reference in New Issue
Block a user