Uh oh!
There was an error while loading.Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork966
Expand file tree
/
Copy pathutil.py
More file actions
121 lines (85 loc) · 3.59 KB
/
util.py
File metadata and controls
121 lines (85 loc) · 3.59 KB
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
# This module is part of GitPython and is released under the
# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
"""Index utilities."""
__all__= ["TemporaryFileSwap","post_clear_cache","default_index","git_working_dir"]
importcontextlib
fromfunctoolsimportwraps
importos
importos.pathasosp
importstruct
importtempfile
fromtypesimportTracebackType
# typing ----------------------------------------------------------------------
fromtypingimportAny,Callable,TYPE_CHECKING,Optional,Type,cast
fromgit.typesimportLiteral,PathLike,_T
ifTYPE_CHECKING:
fromgit.indeximportIndexFile
# ---------------------------------------------------------------------------------
# { Aliases
pack=struct.pack
unpack=struct.unpack
# } END aliases
classTemporaryFileSwap:
"""Utility class moving a file to a temporary location within the same directory and
moving it back on to where on object deletion."""
__slots__= ("file_path","tmp_file_path")
def__init__(self,file_path:PathLike)->None:
self.file_path=file_path
dirname,basename=osp.split(file_path)
fd,self.tmp_file_path=tempfile.mkstemp(prefix=basename,dir=dirname)
os.close(fd)
withcontextlib.suppress(OSError):# It may be that the source does not exist.
os.replace(self.file_path,self.tmp_file_path)
def__enter__(self)->"TemporaryFileSwap":
returnself
def__exit__(
self,
exc_type:Optional[Type[BaseException]],
exc_val:Optional[BaseException],
exc_tb:Optional[TracebackType],
)->Literal[False]:
ifosp.isfile(self.tmp_file_path):
os.replace(self.tmp_file_path,self.file_path)
returnFalse
# { Decorators
defpost_clear_cache(func:Callable[...,_T])->Callable[...,_T]:
"""Decorator for functions that alter the index using the git command.
When a git command alters the index, this invalidates our possibly existing entries
dictionary, which is why it must be deleted to allow it to be lazily reread later.
"""
@wraps(func)
defpost_clear_cache_if_not_raised(self:"IndexFile",*args:Any,**kwargs:Any)->_T:
rval=func(self,*args,**kwargs)
self._delete_entries_cache()
returnrval
# END wrapper method
returnpost_clear_cache_if_not_raised
defdefault_index(func:Callable[...,_T])->Callable[...,_T]:
"""Decorator ensuring the wrapped method may only run if we are the default
repository index.
This is as we rely on git commands that operate on that index only.
"""
@wraps(func)
defcheck_default_index(self:"IndexFile",*args:Any,**kwargs:Any)->_T:
ifself._file_path!=self._index_path():
raiseAssertionError(
"Cannot call %r on indices that do not represent the default git index"%func.__name__
)
returnfunc(self,*args,**kwargs)
# END wrapper method
returncheck_default_index
defgit_working_dir(func:Callable[...,_T])->Callable[...,_T]:
"""Decorator which changes the current working dir to the one of the git
repository in order to ensure relative paths are handled correctly."""
@wraps(func)
defset_git_working_dir(self:"IndexFile",*args:Any,**kwargs:Any)->_T:
cur_wd=os.getcwd()
os.chdir(cast(PathLike,self.repo.working_tree_dir))
try:
returnfunc(self,*args,**kwargs)
finally:
os.chdir(cur_wd)
# END handle working dir
# END wrapper
returnset_git_working_dir
# } END decorators