github/git-batch
0
0
Fork 0
mirror of https://github.com/thegeeklab/git-batch.git synced 2024-11-01 01:00:42 +00:00
Code Issues Projects Releases Wiki Activity

feat: add option to fetch single directory from branch (#146)

Browse Source 
BREAKING CHANGE: Environment variable `GIT_BATCH_IGNORE_EXISTING_REPO` was renamed to `GIT_BATCH_IGNORE_EXISTING_REPO`.

BREAKING CHANGE: If a line in the `batchfile` does not contain a branch, `main` s used as default instead of `master`.
This commit is contained in:
Robert Kaussow 2022-05-25 23:20:49 +02:00 committed by GitHub
parent 5712755dba
commit 00b969fdec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 65 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
gitbatch/cli.py
View File

@ -3,7 +3,10 @@
import argparse
import os
import shutil
import tempfile
from collections import defaultdict
from pathlib import Path
from urllib.parse import urlparse
import git
@ -55,7 +58,7 @@ class GitBatch:
input_file_raw = os.environ.get("GIT_BATCH_INPUT_FILE", ".batchfile")
config["input_file"] = normalize_path(input_file_raw)
config["ignore_existing"] = to_bool(os.environ.get("GIT_BATCH_IGNORE_EXISTING_REPO", True))
config["ignore_existing"] = to_bool(os.environ.get("GIT_BATCH_IGNORE_EXISTING", True))
config["ignore_missing"] = to_bool(os.environ.get("GIT_BATCH_IGNORE_MISSING_REMOTE", True))
return config
@ -68,7 +71,14 @@ class GitBatch:
line = line.strip()
if line and not line.startswith("#"):
try:
url, branch, dest = [x.strip() for x in line.split(";")]
url, src, dest = [x.strip() for x in line.split(";")]
branch, *_ = [x.strip() for x in src.split(":")]
path = None
if len(_) > 0:
path = Path(_[0])
path = path.relative_to(path.anchor)
except ValueError as e:
self.log.sysexit_with_message(
"Wrong numer of delimiters in line {line_num}: {exp}".format(
@ -80,7 +90,8 @@ class GitBatch:
url_parts = urlparse(url)
repo["url"] = url
repo["branch"] = branch or "master"
repo["branch"] = branch or "main"
repo["path"] = path
repo["name"] = os.path.basename(url_parts.path)
repo["rel_dest"] = dest
repo["dest"] = normalize_path(dest) or normalize_path(
@ -94,36 +105,66 @@ class GitBatch:
)
return repos
def _repos_clone(self, repos, ignore_existing):
def _repos_clone(self, repos):
for repo in repos:
with tempfile.TemporaryDirectory(prefix="gitbatch_") as tmp:
try:
options = ["--branch={}".format(repo["branch"]), "--single-branch"]
git.Repo.clone_from(repo["url"], repo["dest"], multi_options=options)
git.Repo.clone_from(repo["url"], tmp, multi_options=options)
os.makedirs(repo["dest"], 0o750, self.config["ignore_existing"])
except git.exc.GitCommandError as e:
passed = False
skip = False
err_raw = e.stderr.strip().splitlines()[:-1]
err = [
x.split(":", 1)[1].strip().replace(repo["dest"], repo["rel_dest"])
for x in err_raw
]
if any(["already exists and is not an empty directory" in item for item in err]):
if self.config["ignore_existing"]:
self.logger.warn("Git error: {}".format("\n".join(err)))
passed = True
if any(["Could not find remote branch" in item for item in err]):
if any(["could not find remote branch" in item for item in err]):
if self.config["ignore_missing"]:
passed = True
skip = True
if not skip:
self.log.sysexit_with_message("Error: {}".format("\n".join(err)))
except FileExistsError:
self._file_exist_handler()
if not passed:
self.log.sysexit_with_message("Git error: {}".format("\n".join(err)))
try:
path = tmp
if repo["path"]:
path = normalize_path(os.path.join(tmp, repo["path"]))
if not os.path.isdir(path):
raise FileNotFoundError(Path(path).relative_to(tmp))
shutil.copytree(
path,
repo["dest"],
ignore=shutil.ignore_patterns(".git"),
dirs_exist_ok=self.config["ignore_existing"]
)
except FileExistsError:
self._file_exist_handler()
except FileNotFoundError as e:
self.log.sysexit_with_message(
"Error: directory '{}' not found in repository '{}'".format(
e, repo["name"]
)
)
def _file_exist_handler(self):
skip = False
err = ["direcory already exists"]
if self.config["ignore_existing"]:
self.logger.warn("Error: {}".format("\n".join(err)))
skip = True
if not skip:
self.log.sysexit_with_message("Error: {}".format("\n".join(err)))
def run(self):
self.log.set_level(self.config["logging"]["level"])
if os.path.isfile(self.config["input_file"]):
repos = self._repos_from_file(self.config["input_file"])
self._repos_clone(repos, self.config["ignore_existing"])
self._repos_clone(repos)
else:
self.log.sysexit_with_message(
"The given batch file at '{}' does not exist".format(