Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/db.py: 84%

1# This module is part of GitPython and is released under the 

2# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/ 

3 

4"""Module with our own gitdb implementation - it uses the git command.""" 

5 

6__all__ = ["GitCmdObjectDB", "GitDB"] 

7 

8from subprocess import PIPE 

9 

10from gitdb.base import IStream, OInfo, OStream 

11from gitdb.db import GitDB, LooseObjectDB 

12from gitdb.exc import BadObject 

13from gitdb.fun import stream_copy 

14 

15from git.compat import force_text 

16from git.util import bin_to_hex, hex_to_bin 

17from git.exc import GitCommandError 

18 

19# typing------------------------------------------------- 

20 

21from typing import TYPE_CHECKING 

22 

23from git.types import PathLike 

24 

25if TYPE_CHECKING: 

26 from git.cmd import Git 

27 

28# -------------------------------------------------------- 

29 

30 

31class GitCmdObjectDB(LooseObjectDB): 

32 """A database representing the default git object store, which includes loose 

33 objects, pack files and an alternates file. 

34 

35 It will create objects only in the loose object database. 

36 """ 

37 

38 def __init__(self, root_path: PathLike, git: "Git") -> None: 

39 """Initialize this instance with the root and a git command.""" 

40 super().__init__(root_path) 

41 self._git = git 

42 

43 def info(self, binsha: bytes) -> OInfo: 

44 """Get a git object header (using git itself).""" 

45 hexsha, typename, size = self._git.get_object_header(bin_to_hex(binsha)) 

46 return OInfo(hex_to_bin(hexsha), typename, size) 

47 

48 def stream(self, binsha: bytes) -> OStream: 

49 """Get git object data as a stream supporting ``read()`` (using git itself).""" 

50 hexsha, typename, size, stream = self._git.stream_object_data(bin_to_hex(binsha)) 

51 return OStream(hex_to_bin(hexsha), typename, size, stream) 

52 

53 def store(self, istream: IStream) -> IStream: 

54 """Store an object using git itself.""" 

55 if istream.binsha is not None or self.ostream() is not None: 

56 return super().store(istream) 

57 

58 proc = self._git.hash_object( 

59 "-t", force_text(istream.type), "-w", "--stdin", "--literally", as_process=True, istream=PIPE 

60 ) 

61 assert proc.stdin is not None 

62 try: 

63 stream_copy(istream.read, proc.stdin.write, istream.size, self.stream_chunk_size) 

64 finally: 

65 proc.stdin.close() 

66 assert proc.stdout is not None 

67 hexsha = proc.stdout.read().strip() 

68 proc.wait() 

69 istream.binsha = hex_to_bin(hexsha) 

70 return istream 

71 

72 # { Interface 

73 

74 def partial_to_complete_sha_hex(self, partial_hexsha: str) -> bytes: 

75 """ 

76 :return: 

77 Full binary 20 byte sha from the given partial hexsha 

78 

79 :raise gitdb.exc.AmbiguousObjectName: 

80 

81 :raise gitdb.exc.BadObject: 

82 

83 :note: 

84 Currently we only raise :exc:`~gitdb.exc.BadObject` as git does not 

85 communicate ambiguous objects separately. 

86 """ 

87 try: 

88 hexsha, _typename, _size = self._git.get_object_header(partial_hexsha) 

89 return hex_to_bin(hexsha) 

90 except (GitCommandError, ValueError) as e: 

91 raise BadObject(partial_hexsha) from e 

92 # END handle exceptions 

93 

94 # } END interface