/
repository.py
1548 lines (1291 loc) · 54.1 KB
/
repository.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import atexit
import os
import re
import subprocess
import tempfile
import threading
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Callable, Dict, Iterator, List, Optional, Tuple, Union
from urllib.parse import urlparse
from tqdm.auto import tqdm
from huggingface_hub.constants import REPO_TYPES_URL_PREFIXES, REPOCARD_NAME
from huggingface_hub.repocard import metadata_load, metadata_save
from requests.exceptions import HTTPError
from .hf_api import HfApi, HfFolder, repo_type_and_id_from_hf_id
from .lfs import LFS_MULTIPART_UPLOAD_COMMAND
from .utils import logging, run_subprocess
logger = logging.get_logger(__name__)
class CommandInProgress:
"""
Utility to follow commands launched asynchronously.
"""
def __init__(
self,
title: str,
is_done_method: Callable,
status_method: Callable,
process: subprocess.Popen,
post_method: Optional[Callable] = None,
):
self.title = title
self._is_done = is_done_method
self._status = status_method
self._process = process
self._stderr = ""
self._stdout = ""
self._post_method = post_method
@property
def is_done(self) -> bool:
"""
Whether the process is done.
"""
result = self._is_done()
if result and self._post_method is not None:
self._post_method()
self._post_method = None
return result
@property
def status(self) -> int:
"""
The exit code/status of the current action. Will return `0` if the
command has completed successfully, and a number between 1 and 255 if
the process errored-out.
Will return -1 if the command is still ongoing.
"""
return self._status()
@property
def failed(self) -> bool:
"""
Whether the process errored-out.
"""
return self.status > 0
@property
def stderr(self) -> str:
"""
The current output message on the standard error.
"""
self._stderr += self._process.stderr.read()
return self._stderr
@property
def stdout(self) -> str:
"""
The current output message on the standard output.
"""
self._stdout += self._process.stdout.read()
return self._stdout
def __repr__(self):
status = self.status
if status == -1:
status = "running"
return (
f"[{self.title} command, status code: {status},"
f" {'in progress.' if not self.is_done else 'finished.'} PID:"
f" {self._process.pid}]"
)
def is_git_repo(folder: Union[str, Path]) -> bool:
"""
Check if the folder is the root or part of a git repository
Args:
folder (`str`):
The folder in which to run the command.
Returns:
`bool`: `True` if the repository is part of a repository, `False`
otherwise.
"""
folder_exists = os.path.exists(os.path.join(folder, ".git"))
git_branch = subprocess.run(
"git branch".split(), cwd=folder, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
return folder_exists and git_branch.returncode == 0
def is_local_clone(folder: Union[str, Path], remote_url: str) -> bool:
"""
Check if the folder is a local clone of the remote_url
Args:
folder (`str` or `Path`):
The folder in which to run the command.
remote_url (`str`):
The url of a git repository.
Returns:
`bool`: `True` if the repository is a local clone of the remote
repository specified, `False` otherwise.
"""
if not is_git_repo(folder):
return False
remotes = run_subprocess("git remote -v".split(), folder).stdout
# Remove token for the test with remotes.
remote_url = re.sub(r"https://.*@", "https://", remote_url)
remotes = [re.sub(r"https://.*@", "https://", remote) for remote in remotes.split()]
return remote_url in remotes
def is_tracked_with_lfs(filename: Union[str, Path]) -> bool:
"""
Check if the file passed is tracked with git-lfs.
Args:
filename (`str` or `Path`):
The filename to check.
Returns:
`bool`: `True` if the file passed is tracked with git-lfs, `False`
otherwise.
"""
folder = Path(filename).parent
filename = Path(filename).name
try:
p = run_subprocess("git check-attr -a".split() + [filename], folder)
attributes = p.stdout.strip()
except subprocess.CalledProcessError as exc:
if not is_git_repo(folder):
return False
else:
raise OSError(exc.stderr)
if len(attributes) == 0:
return False
found_lfs_tag = {"diff": False, "merge": False, "filter": False}
for attribute in attributes.split("\n"):
for tag in found_lfs_tag.keys():
if tag in attribute and "lfs" in attribute:
found_lfs_tag[tag] = True
return all(found_lfs_tag.values())
def is_git_ignored(filename: Union[str, Path]) -> bool:
"""
Check if file is git-ignored. Supports nested .gitignore files.
Args:
filename (`str` or `Path`):
The filename to check.
Returns:
`bool`: `True` if the file passed is ignored by `git`, `False`
otherwise.
"""
folder = Path(filename).parent
filename = Path(filename).name
try:
p = run_subprocess("git check-ignore".split() + [filename], folder, check=False)
# Will return exit code 1 if not gitignored
is_ignored = not bool(p.returncode)
except subprocess.CalledProcessError as exc:
raise OSError(exc.stderr)
return is_ignored
def is_binary_file(filename: Union[str, Path]) -> bool:
"""
Check if file is a binary file.
Args:
filename (`str` or `Path`):
The filename to check.
Returns:
`bool`: `True` if the file passed is a binary file, `False` otherwise.
"""
try:
with open(filename, "rb") as f:
content = f.read(10 * (1024**2)) # Read a maximum of 10MB
# Code sample taken from the following stack overflow thread
# https://stackoverflow.com/questions/898669/how-can-i-detect-if-a-file-is-binary-non-text-in-python/7392391#7392391
text_chars = bytearray(
{7, 8, 9, 10, 12, 13, 27} | set(range(0x20, 0x100)) - {0x7F}
)
return bool(content.translate(None, text_chars))
except UnicodeDecodeError:
return True
def files_to_be_staged(pattern: str, folder: Union[str, Path]) -> List[str]:
"""
Returns a list of filenames that are to be staged.
Args:
pattern (`str` or `Path`):
The pattern of filenames to check. Put `.` to get all files.
folder (`str` or `Path`):
The folder in which to run the command.
Returns:
`List[str]`: List of files that are to be staged.
"""
try:
p = run_subprocess("git ls-files -mo".split() + [pattern], folder)
if len(p.stdout.strip()):
files = p.stdout.strip().split("\n")
else:
files = []
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
return files
def is_tracked_upstream(folder: Union[str, Path]) -> bool:
"""
Check if the current checked-out branch is tracked upstream.
Args:
folder (`str` or `Path`):
The folder in which to run the command.
Returns:
`bool`: `True` if the current checked-out branch is tracked upstream,
`False` otherwise.
"""
try:
run_subprocess(
"git rev-parse --symbolic-full-name --abbrev-ref @{u}".split(), folder
)
return True
except subprocess.CalledProcessError as exc:
if "HEAD" in exc.stderr:
raise OSError("No branch checked out")
return False
def commits_to_push(folder: Union[str, Path], upstream: Optional[str] = None) -> int:
"""
Check the number of commits that would be pushed upstream
Args:
folder (`str` or `Path`):
The folder in which to run the command.
upstream (`str`, *optional*):
The name of the upstream repository with which the comparison should be
made.
Returns:
`int`: Number of commits that would be pushed upstream were a `git
push` to proceed.
"""
try:
result = run_subprocess(f"git cherry -v {upstream or ''}".split(), folder)
return len(result.stdout.split("\n")) - 1
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
@contextmanager
def _lfs_log_progress():
"""
This is a context manager that will log the Git LFS progress of cleaning,
smudging, pulling and pushing.
"""
if logger.getEffectiveLevel() >= logging.ERROR:
try:
yield
finally:
return
def output_progress(stopping_event: threading.Event):
"""
To be launched as a separate thread with an event meaning it should stop
the tail.
"""
pbars = {}
def close_pbars():
for pbar in pbars.values():
pbar["bar"].update(pbar["bar"].total - pbar["past_bytes"])
pbar["bar"].refresh()
pbar["bar"].close()
def tail_file(filename) -> Iterator[str]:
"""
Creates a generator to be iterated through, which will return each
line one by one. Will stop tailing the file if the stopping_event is
set.
"""
with open(filename, "r") as file:
current_line = ""
while True:
if stopping_event.is_set():
close_pbars()
break
line_bit = file.readline()
if line_bit is not None and not len(line_bit.strip()) == 0:
current_line += line_bit
if current_line.endswith("\n"):
yield current_line
current_line = ""
else:
time.sleep(1)
# If the file isn't created yet, wait for a few seconds before trying again.
# Can be interrupted with the stopping_event.
while not os.path.exists(os.environ["GIT_LFS_PROGRESS"]):
if stopping_event.is_set():
close_pbars()
return
time.sleep(2)
for line in tail_file(os.environ["GIT_LFS_PROGRESS"]):
state, file_progress, byte_progress, filename = line.split()
description = f"{state.capitalize()} file {filename}"
current_bytes, total_bytes = byte_progress.split("/")
current_bytes = int(current_bytes)
total_bytes = int(total_bytes)
if pbars.get((state, filename)) is None:
pbars[(state, filename)] = {
"bar": tqdm(
desc=description,
initial=current_bytes,
total=total_bytes,
unit="B",
unit_scale=True,
unit_divisor=1024,
),
"past_bytes": current_bytes,
}
else:
past_bytes = pbars[(state, filename)]["past_bytes"]
pbars[(state, filename)]["bar"].update(current_bytes - past_bytes)
pbars[(state, filename)]["past_bytes"] = current_bytes
current_lfs_progress_value = os.environ.get("GIT_LFS_PROGRESS", "")
with tempfile.TemporaryDirectory() as tmpdir:
os.environ["GIT_LFS_PROGRESS"] = os.path.join(tmpdir, "lfs_progress")
logger.debug(f"Following progress in {os.environ['GIT_LFS_PROGRESS']}")
exit_event = threading.Event()
x = threading.Thread(target=output_progress, args=(exit_event,), daemon=True)
x.start()
try:
yield
finally:
exit_event.set()
x.join()
os.environ["GIT_LFS_PROGRESS"] = current_lfs_progress_value
class Repository:
"""
Helper class to wrap the git and git-lfs commands.
The aim is to facilitate interacting with huggingface.co hosted model or
dataset repos, though not a lot here (if any) is actually specific to
huggingface.co.
"""
command_queue: List[CommandInProgress]
def __init__(
self,
local_dir: str,
clone_from: Optional[str] = None,
repo_type: Optional[str] = None,
use_auth_token: Union[bool, str] = True,
git_user: Optional[str] = None,
git_email: Optional[str] = None,
revision: Optional[str] = None,
private: bool = False,
skip_lfs_files: bool = False,
client: Optional[HfApi] = None,
):
"""
Instantiate a local clone of a git repo.
If specifying a `clone_from`, it will clone an existing remote
repository, for instance one that was previously created using
`HfApi().create_repo(repo_id=repo_name)`.
`Repository` uses the local git credentials by default, but if required,
the `huggingface_token` as well as the git `user` and the `email` can be
explicitly specified.
Args:
local_dir (`str`):
path (e.g. `'my_trained_model/'`) to the local directory, where
the `Repository` will be initalized.
clone_from (`str`, *optional*):
repository url (e.g.
`'https://huggingface.co/philschmid/playground-tests'`).
repo_type (`str`, *optional*):
To set when creating a repo: et to "dataset" or "space" if
creating a dataset or space, default is model.
use_auth_token (`str` or `bool`, *optional*, defaults to `True`):
huggingface_token can be extract from `HfApi().login(username,
password)` and is used to authenticate against the hub (useful
from Google Colab for instance).
git_user (`str`, *optional*):
will override the `git config user.name` for committing and
pushing files to the hub.
git_email (`str`, *optional*):
will override the `git config user.email` for committing and
pushing files to the hub.
revision (`str`, *optional*):
Revision to checkout after initializing the repository. If the
revision doesn't exist, a branch will be created with that
revision name from the default branch's current HEAD.
private (`bool`, *optional*, defaults to `False`):
whether the repository is private or not.
skip_lfs_files (`bool`, *optional*, defaults to `False`):
whether to skip git-LFS files or not.
client (`HfApi`, *optional*):
Instance of HfApi to use when calling the HF Hub API. A new
instance will be created if this is left to `None`.
"""
os.makedirs(local_dir, exist_ok=True)
self.local_dir = os.path.join(os.getcwd(), local_dir)
self.repo_type = repo_type
self.command_queue = []
self.private = private
self.skip_lfs_files = skip_lfs_files
self.client = client if client is not None else HfApi()
self.check_git_versions()
if isinstance(use_auth_token, str):
self.huggingface_token = use_auth_token
elif use_auth_token:
self.huggingface_token = HfFolder.get_token()
else:
self.huggingface_token = None
if clone_from is not None:
self.clone_from(repo_url=clone_from)
else:
if is_git_repo(self.local_dir):
logger.debug("[Repository] is a valid git repo")
else:
raise ValueError(
"If not specifying `clone_from`, you need to pass Repository a"
" valid git clone."
)
if self.huggingface_token is not None and (
git_email is None or git_user is None
):
user = self.client.whoami(self.huggingface_token)
if git_email is None:
git_email = user["email"]
if git_user is None:
git_user = user["fullname"]
if git_user is not None or git_email is not None:
self.git_config_username_and_email(git_user, git_email)
self.lfs_enable_largefiles()
self.git_credential_helper_store()
if revision is not None:
self.git_checkout(revision, create_branch_ok=True)
# This ensures that all commands exit before exiting the Python runtime.
# This will ensure all pushes register on the hub, even if other errors happen in subsequent operations.
atexit.register(self.wait_for_commands)
@property
def current_branch(self) -> str:
"""
Returns the current checked out branch.
Returns:
`str`: Current checked out branch.
"""
try:
result = run_subprocess(
"git rev-parse --abbrev-ref HEAD".split(), self.local_dir
).stdout.strip()
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
return result
def check_git_versions(self):
"""
Checks that `git` and `git-lfs` can be run.
Raises:
`EnvironmentError`: if `git` or `git-lfs` are not installed.
"""
try:
git_version = run_subprocess(
"git --version".split(), self.local_dir
).stdout.strip()
except FileNotFoundError:
raise EnvironmentError(
"Looks like you do not have git installed, please install."
)
try:
lfs_version = run_subprocess(
"git-lfs --version".split(), self.local_dir
).stdout.strip()
except FileNotFoundError:
raise EnvironmentError(
"Looks like you do not have git-lfs installed, please install."
" You can install from https://git-lfs.github.com/."
" Then run `git lfs install` (you only have to do this once)."
)
logger.info(git_version + "\n" + lfs_version)
def clone_from(self, repo_url: str, use_auth_token: Union[bool, str, None] = None):
"""
Clone from a remote. If the folder already exists, will try to clone the
repository within it.
If this folder is a git repository with linked history, will try to
update the repository.
Args:
repo_url (`str`):
The URL from which to clone the repository
use_auth_token (`Union[str, bool]`, *optional*):
Whether to use the authentication token. It can be:
- a string which is the token itself
- `False`, which would not use the authentication token
- `True`, which would fetch the authentication token from the
local folder and use it (you should be logged in for this to
work).
- `None`, which would retrieve the value of
`self.huggingface_token`.
Raises:
`ValueError`: if the `token` cannot be identified and the `private`
keyword is set to `True`. The `token`
must be passed in order to handle private repositories.
Raises:
`EnvironmentError`: if you are trying to clone the repository in a
non-empty folder, or if the `git`
operations raise errors.
"""
token = use_auth_token if use_auth_token is not None else self.huggingface_token
if token is None and self.private:
raise ValueError(
"Couldn't load Hugging Face Authorization Token. Credentials are"
" required to work with private repositories. Please login in using"
" `huggingface-cli login` or provide your token manually with the"
" `use_auth_token` key."
)
hub_url = self.client.endpoint
if hub_url in repo_url or (
"http" not in repo_url and len(repo_url.split("/")) <= 2
):
repo_type, namespace, repo_id = repo_type_and_id_from_hf_id(
repo_url, hub_url=hub_url
)
if repo_type is not None:
self.repo_type = repo_type
repo_url = hub_url + "/"
if self.repo_type in REPO_TYPES_URL_PREFIXES:
repo_url += REPO_TYPES_URL_PREFIXES[self.repo_type]
if token is not None:
whoami_info = self.client.whoami(token)
user = whoami_info["name"]
valid_organisations = [org["name"] for org in whoami_info["orgs"]]
if namespace is not None:
repo_id = f"{namespace}/{repo_id}"
repo_url += repo_id
scheme = urlparse(repo_url).scheme
repo_url = repo_url.replace(f"{scheme}://", f"{scheme}://user:{token}@")
if namespace == user or namespace in valid_organisations:
try:
_ = HfApi().repo_info(
f"{repo_id}", repo_type=self.repo_type, token=token
)
except HTTPError:
if self.repo_type == "space":
raise ValueError(
"Creating a Space through passing Space link to"
" clone_from is not allowed. Make sure the Space exists"
" on Hugging Face Hub."
)
else:
self.client.create_repo(
repo_id=repo_id,
token=token,
repo_type=self.repo_type,
exist_ok=True,
private=self.private,
)
else:
if namespace is not None:
repo_url += f"{namespace}/"
repo_url += repo_id
# For error messages, it's cleaner to show the repo url without the token.
clean_repo_url = re.sub(r"(https?)://.*@", r"\1://", repo_url)
try:
run_subprocess("git lfs install".split(), self.local_dir)
# checks if repository is initialized in a empty repository or in one with files
if len(os.listdir(self.local_dir)) == 0:
logger.warning(f"Cloning {clean_repo_url} into local empty directory.")
with _lfs_log_progress():
env = os.environ.copy()
if self.skip_lfs_files:
env.update({"GIT_LFS_SKIP_SMUDGE": "1"})
run_subprocess(
f"{'git clone' if self.skip_lfs_files else 'git lfs clone'} {repo_url} ."
.split(),
self.local_dir,
env=env,
)
else:
# Check if the folder is the root of a git repository
in_repository = is_git_repo(self.local_dir)
if in_repository:
if is_local_clone(self.local_dir, repo_url):
logger.warning(
f"{self.local_dir} is already a clone of {clean_repo_url}."
" Make sure you pull the latest changes with"
" `repo.git_pull()`."
)
else:
output = run_subprocess(
"git remote get-url origin".split(),
self.local_dir,
check=False,
)
error_msg = (
f"Tried to clone {clean_repo_url} in an unrelated git"
" repository.\nIf you believe this is an error, please add"
f" a remote with the following URL: {clean_repo_url}."
)
if output.returncode == 0:
clean_local_remote_url = re.sub(
r"https://.*@", "https://", output.stdout
)
error_msg += (
"\nLocal path has its origin defined as:"
f" {clean_local_remote_url}"
)
raise EnvironmentError(error_msg)
if not in_repository:
raise EnvironmentError(
"Tried to clone a repository in a non-empty folder that isn't a"
" git repository. If you really want to do this, do it"
" manually:\ngit init && git remote add origin && git pull"
" origin main\n or clone repo to a new folder and move your"
" existing files there afterwards."
)
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
def git_config_username_and_email(
self, git_user: Optional[str] = None, git_email: Optional[str] = None
):
"""
Sets git username and email (only in the current repo).
Args:
git_user (`str`, *optional*):
The username to register through `git`.
git_email (`str`, *optional*):
The email to register through `git`.
"""
try:
if git_user is not None:
run_subprocess(
"git config user.name".split() + [git_user], self.local_dir
)
if git_email is not None:
run_subprocess(
f"git config user.email {git_email}".split(), self.local_dir
)
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
def git_credential_helper_store(self):
"""
Sets the git credential helper to `store`
"""
try:
run_subprocess("git config credential.helper store".split(), self.local_dir)
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
def git_head_hash(self) -> str:
"""
Get commit sha on top of HEAD.
Returns:
`str`: The current checked out commit SHA.
"""
try:
p = run_subprocess("git rev-parse HEAD".split(), self.local_dir)
return p.stdout.strip()
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
def git_remote_url(self) -> str:
"""
Get URL to origin remote.
Returns:
`str`: The URL of the `origin` remote.
"""
try:
p = run_subprocess(
"git config --get remote.origin.url".split(), self.local_dir
)
url = p.stdout.strip()
# Strip basic auth info.
return re.sub(r"https://.*@", "https://", url)
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
def git_head_commit_url(self) -> str:
"""
Get URL to last commit on HEAD. We assume it's been pushed, and the url
scheme is the same one as for GitHub or HuggingFace.
Returns:
`str`: The URL to the current checked-out commit.
"""
sha = self.git_head_hash()
url = self.git_remote_url()
if url.endswith("/"):
url = url[:-1]
return f"{url}/commit/{sha}"
def list_deleted_files(self) -> List[str]:
"""
Returns a list of the files that are deleted in the working directory or
index.
Returns:
`List[str]`: A list of files that have been deleted in the working
directory or index.
"""
try:
git_status = run_subprocess(
"git status -s".split(), self.local_dir
).stdout.strip()
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
if len(git_status) == 0:
return []
# Receives a status like the following
# D .gitignore
# D new_file.json
# AD new_file1.json
# ?? new_file2.json
# ?? new_file4.json
# Strip each line of whitespaces
modified_files_statuses = [status.strip() for status in git_status.split("\n")]
# Only keep files that are deleted using the D prefix
deleted_files_statuses = [
status for status in modified_files_statuses if "D" in status.split()[0]
]
# Remove the D prefix and strip to keep only the relevant filename
deleted_files = [
status.split()[-1].strip() for status in deleted_files_statuses
]
return deleted_files
def lfs_track(
self, patterns: Union[str, List[str]], filename: Optional[bool] = False
):
"""
Tell git-lfs to track files according to a pattern.
Setting the `filename` argument to `True` will treat the arguments as
literal filenames, not as patterns. Any special glob characters in the
filename will be escaped when writing to the `.gitattributes` file.
Args:
patterns (`Union[str, List[str]]`):
The pattern, or list of patterns, to track with git-lfs.
filename (`bool`, *optional*, defaults to `False`):
Whether to use the patterns as literal filenames.
"""
if isinstance(patterns, str):
patterns = [patterns]
try:
for pattern in patterns:
cmd = f"git lfs track {'--filename' if filename else ''}".split()
cmd.append(pattern)
run_subprocess(cmd, self.local_dir)
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
def lfs_untrack(self, patterns: Union[str, List[str]]):
"""
Tell git-lfs to untrack those files.
Args:
patterns (`Union[str, List[str]]`):
The pattern, or list of patterns, to untrack with git-lfs.
"""
if isinstance(patterns, str):
patterns = [patterns]
try:
for pattern in patterns:
run_subprocess("git lfs untrack".split() + [pattern], self.local_dir)
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
def lfs_enable_largefiles(self):
"""
HF-specific. This enables upload support of files >5GB.
"""
try:
lfs_config = "git config lfs.customtransfer.multipart"
run_subprocess(f"{lfs_config}.path huggingface-cli".split(), self.local_dir)
run_subprocess(
f"{lfs_config}.args {LFS_MULTIPART_UPLOAD_COMMAND}".split(),
self.local_dir,
)
except subprocess.CalledProcessError as exc:
raise EnvironmentError(exc.stderr)
def auto_track_binary_files(self, pattern: Optional[str] = ".") -> List[str]:
"""
Automatically track binary files with git-lfs.
Args:
pattern (`str`, *optional*, defaults to "."):
The pattern with which to track files that are binary.
Returns:
`List[str]`: List of filenames that are now tracked due to being
binary files
"""
files_to_be_tracked_with_lfs = []
deleted_files = self.list_deleted_files()
for filename in files_to_be_staged(pattern, folder=self.local_dir):
if filename in deleted_files:
continue
path_to_file = os.path.join(os.getcwd(), self.local_dir, filename)
if not (is_tracked_with_lfs(path_to_file) or is_git_ignored(path_to_file)):
size_in_mb = os.path.getsize(path_to_file) / (1024 * 1024)
if size_in_mb >= 10:
logger.warning(
"Parsing a large file to check if binary or not. Tracking large"
" files using `repository.auto_track_large_files` is"
" recommended so as to not load the full file in memory."
)
is_binary = is_binary_file(path_to_file)
if is_binary:
self.lfs_track(filename)
files_to_be_tracked_with_lfs.append(filename)
# Cleanup the .gitattributes if files were deleted
self.lfs_untrack(deleted_files)
return files_to_be_tracked_with_lfs
def auto_track_large_files(self, pattern: Optional[str] = ".") -> List[str]:
"""
Automatically track large files (files that weigh more than 10MBs) with
git-lfs.
Args:
pattern (`str`, *optional*, defaults to "."):
The pattern with which to track files that are above 10MBs.
Returns:
`List[str]`: List of filenames that are now tracked due to their
size.
"""
files_to_be_tracked_with_lfs = []
deleted_files = self.list_deleted_files()
for filename in files_to_be_staged(pattern, folder=self.local_dir):
if filename in deleted_files:
continue
path_to_file = os.path.join(os.getcwd(), self.local_dir, filename)
size_in_mb = os.path.getsize(path_to_file) / (1024 * 1024)
if (
size_in_mb >= 10
and not is_tracked_with_lfs(path_to_file)
and not is_git_ignored(path_to_file)
):
self.lfs_track(filename)
files_to_be_tracked_with_lfs.append(filename)
# Cleanup the .gitattributes if files were deleted
self.lfs_untrack(deleted_files)
return files_to_be_tracked_with_lfs
def lfs_prune(self, recent=False):
"""
git lfs prune
Args:
recent (`bool`, *optional*, defaults to `False`):
Whether to prune files even if they were referenced by recent
commits. See the following