Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def move_in_pack(self, path):
"""Move a specific file containing a pack into the pack directory.
:note: The file should be on the same file system as the
packs directory.
:param path: Path to the pack file.
"""
with PackData(path) as p:
entries = p.sorted_entries()
basename = self._get_pack_basepath(entries)
with GitFile(basename+".idx", "wb") as f:
write_pack_index_v2(f, entries, p.get_stored_checksum())
os.rename(path, basename + ".pack")
final_pack = Pack(basename)
self._add_known_pack(basename, final_pack)
return final_pack
def keep(self, msg=None):
"""Add a .keep file for the pack, preventing git from garbage collecting it.
:param msg: A message written inside the .keep file; can be used later
to determine whether or not a .keep file is obsolete.
:return: The path of the .keep file, as a string.
"""
keepfile_name = '%s.keep' % self._basename
with GitFile(keepfile_name, 'wb') as keepfile:
if msg:
keepfile.write(msg)
keepfile.write(b'\n')
return keepfile_name
def remove_if_equals(self, name, old_ref):
"""Remove a refname only if it currently equals old_ref.
This method does not follow symbolic references. It can be used to
perform an atomic compare-and-delete operation.
:param name: The refname to delete.
:param old_ref: The old sha the refname must refer to, or None to
delete unconditionally.
:return: True if the delete was successful, False otherwise.
"""
self._check_refname(name)
filename = self.refpath(name)
ensure_dir_exists(os.path.dirname(filename))
f = GitFile(filename, 'wb')
try:
if old_ref is not None:
orig_ref = self.read_loose_ref(name)
if orig_ref is None:
orig_ref = self.get_packed_refs().get(name, ZERO_SHA)
if orig_ref != old_ref:
return False
# may only be packed
try:
os.remove(filename)
except OSError as e:
if e.errno != errno.ENOENT:
raise
self._remove_packed_ref(name)
finally:
# never write, we just wanted the lock
def read(self):
"""Read current contents of index from disk."""
if not os.path.exists(self._filename):
return
f = GitFile(self._filename, 'rb')
try:
f = SHA1Reader(f)
for x in read_index(f):
self[x[0]] = IndexEntry(*x[1:])
# FIXME: Additional data?
f.read(os.path.getsize(self._filename)-f.tell()-20)
f.check_sha()
finally:
f.close()
def move_in_pack(self, path):
"""Move a specific file containing a pack into the pack directory.
Note: The file should be on the same file system as the
packs directory.
Args:
path: Path to the pack file.
"""
with PackData(path) as p:
entries = p.sorted_entries()
basename = self._get_pack_basepath(entries)
index_name = basename + ".idx"
if not os.path.exists(index_name):
with GitFile(index_name, "wb") as f:
write_pack_index_v2(f, entries, p.get_stored_checksum())
for pack in self.packs:
if pack._basename == basename:
return pack
target_pack = basename + '.pack'
if sys.platform == 'win32':
# Windows might have the target pack file lingering. Attempt
# removal, silently passing if the target does not exist.
try:
os.remove(target_pack)
except (IOError, OSError) as e:
if e.errno != errno.ENOENT:
raise
os.rename(path, target_pack)
final_pack = Pack(basename)
self._add_cached_pack(basename, final_pack)
def get_description(self):
"""Retrieve the description of this repository.
:return: A string describing the repository or None.
"""
path = os.path.join(self._controldir, 'description')
try:
with GitFile(path, 'rb') as f:
return f.read()
except (IOError, OSError) as e:
if e.errno != errno.ENOENT:
raise
return None
def write_to_path(self, path=None):
"""Write configuration to a file on disk."""
if path is None:
path = self.path
with GitFile(path, 'wb') as f:
self.write_to_file(f)
def _put_named_file(self, path, contents):
"""Write a file to the control dir with the given name and contents.
:param path: The path to the file, relative to the control dir.
:param contents: A string to write to the file.
"""
path = path.lstrip(os.path.sep)
with GitFile(os.path.join(self.controldir(), path), 'wb') as f:
f.write(contents)
"""Remove a refname only if it currently equals old_ref.
This method does not follow symbolic references. It can be used to
perform an atomic compare-and-delete operation.
Args:
name: The refname to delete.
old_ref: The old sha the refname must refer to, or None to
delete unconditionally.
message: Optional message
Returns: True if the delete was successful, False otherwise.
"""
self._check_refname(name)
filename = self.refpath(name)
ensure_dir_exists(os.path.dirname(filename))
f = GitFile(filename, 'wb')
try:
if old_ref is not None:
orig_ref = self.read_loose_ref(name)
if orig_ref is None:
orig_ref = self.get_packed_refs().get(name, ZERO_SHA)
if orig_ref != old_ref:
return False
# remove the reference file itself
try:
os.remove(filename)
except OSError as e:
if e.errno != errno.ENOENT: # may only be packed
raise
self._remove_packed_ref(name)