Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/gitdb/base.py: 71%
138 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1# Copyright (C) 2010, 2011 Sebastian Thiel (byronimo@gmail.com) and contributors
2#
3# This module is part of GitDB and is released under
4# the New BSD License: http://www.opensource.org/licenses/bsd-license.php
5"""Module with basic data structures - they are designed to be lightweight and fast"""
6from gitdb.util import bin_to_hex
8from gitdb.fun import (
9 type_id_to_type_map,
10 type_to_type_id_map
11)
13__all__ = ('OInfo', 'OPackInfo', 'ODeltaPackInfo',
14 'OStream', 'OPackStream', 'ODeltaPackStream',
15 'IStream', 'InvalidOInfo', 'InvalidOStream')
17#{ ODB Bases
20class OInfo(tuple):
22 """Carries information about an object in an ODB, providing information
23 about the binary sha of the object, the type_string as well as the uncompressed size
24 in bytes.
26 It can be accessed using tuple notation and using attribute access notation::
28 assert dbi[0] == dbi.binsha
29 assert dbi[1] == dbi.type
30 assert dbi[2] == dbi.size
32 The type is designed to be as lightweight as possible."""
33 __slots__ = tuple()
35 def __new__(cls, sha, type, size):
36 return tuple.__new__(cls, (sha, type, size))
38 def __init__(self, *args):
39 tuple.__init__(self)
41 #{ Interface
42 @property
43 def binsha(self):
44 """:return: our sha as binary, 20 bytes"""
45 return self[0]
47 @property
48 def hexsha(self):
49 """:return: our sha, hex encoded, 40 bytes"""
50 return bin_to_hex(self[0])
52 @property
53 def type(self):
54 return self[1]
56 @property
57 def type_id(self):
58 return type_to_type_id_map[self[1]]
60 @property
61 def size(self):
62 return self[2]
63 #} END interface
66class OPackInfo(tuple):
68 """As OInfo, but provides a type_id property to retrieve the numerical type id, and
69 does not include a sha.
71 Additionally, the pack_offset is the absolute offset into the packfile at which
72 all object information is located. The data_offset property points to the absolute
73 location in the pack at which that actual data stream can be found."""
74 __slots__ = tuple()
76 def __new__(cls, packoffset, type, size):
77 return tuple.__new__(cls, (packoffset, type, size))
79 def __init__(self, *args):
80 tuple.__init__(self)
82 #{ Interface
84 @property
85 def pack_offset(self):
86 return self[0]
88 @property
89 def type(self):
90 return type_id_to_type_map[self[1]]
92 @property
93 def type_id(self):
94 return self[1]
96 @property
97 def size(self):
98 return self[2]
100 #} END interface
103class ODeltaPackInfo(OPackInfo):
105 """Adds delta specific information,
106 Either the 20 byte sha which points to some object in the database,
107 or the negative offset from the pack_offset, so that pack_offset - delta_info yields
108 the pack offset of the base object"""
109 __slots__ = tuple()
111 def __new__(cls, packoffset, type, size, delta_info):
112 return tuple.__new__(cls, (packoffset, type, size, delta_info))
114 #{ Interface
115 @property
116 def delta_info(self):
117 return self[3]
118 #} END interface
121class OStream(OInfo):
123 """Base for object streams retrieved from the database, providing additional
124 information about the stream.
125 Generally, ODB streams are read-only as objects are immutable"""
126 __slots__ = tuple()
128 def __new__(cls, sha, type, size, stream, *args, **kwargs):
129 """Helps with the initialization of subclasses"""
130 return tuple.__new__(cls, (sha, type, size, stream))
132 def __init__(self, *args, **kwargs):
133 tuple.__init__(self)
135 #{ Stream Reader Interface
137 def read(self, size=-1):
138 return self[3].read(size)
140 @property
141 def stream(self):
142 return self[3]
144 #} END stream reader interface
147class ODeltaStream(OStream):
149 """Uses size info of its stream, delaying reads"""
151 def __new__(cls, sha, type, size, stream, *args, **kwargs):
152 """Helps with the initialization of subclasses"""
153 return tuple.__new__(cls, (sha, type, size, stream))
155 #{ Stream Reader Interface
157 @property
158 def size(self):
159 return self[3].size
161 #} END stream reader interface
164class OPackStream(OPackInfo):
166 """Next to pack object information, a stream outputting an undeltified base object
167 is provided"""
168 __slots__ = tuple()
170 def __new__(cls, packoffset, type, size, stream, *args):
171 """Helps with the initialization of subclasses"""
172 return tuple.__new__(cls, (packoffset, type, size, stream))
174 #{ Stream Reader Interface
175 def read(self, size=-1):
176 return self[3].read(size)
178 @property
179 def stream(self):
180 return self[3]
181 #} END stream reader interface
184class ODeltaPackStream(ODeltaPackInfo):
186 """Provides a stream outputting the uncompressed offset delta information"""
187 __slots__ = tuple()
189 def __new__(cls, packoffset, type, size, delta_info, stream):
190 return tuple.__new__(cls, (packoffset, type, size, delta_info, stream))
192 #{ Stream Reader Interface
193 def read(self, size=-1):
194 return self[4].read(size)
196 @property
197 def stream(self):
198 return self[4]
199 #} END stream reader interface
202class IStream(list):
204 """Represents an input content stream to be fed into the ODB. It is mutable to allow
205 the ODB to record information about the operations outcome right in this instance.
207 It provides interfaces for the OStream and a StreamReader to allow the instance
208 to blend in without prior conversion.
210 The only method your content stream must support is 'read'"""
211 __slots__ = tuple()
213 def __new__(cls, type, size, stream, sha=None):
214 return list.__new__(cls, (sha, type, size, stream, None))
216 def __init__(self, type, size, stream, sha=None):
217 list.__init__(self, (sha, type, size, stream, None))
219 #{ Interface
220 @property
221 def hexsha(self):
222 """:return: our sha, hex encoded, 40 bytes"""
223 return bin_to_hex(self[0])
225 def _error(self):
226 """:return: the error that occurred when processing the stream, or None"""
227 return self[4]
229 def _set_error(self, exc):
230 """Set this input stream to the given exc, may be None to reset the error"""
231 self[4] = exc
233 error = property(_error, _set_error)
235 #} END interface
237 #{ Stream Reader Interface
239 def read(self, size=-1):
240 """Implements a simple stream reader interface, passing the read call on
241 to our internal stream"""
242 return self[3].read(size)
244 #} END stream reader interface
246 #{ interface
248 def _set_binsha(self, binsha):
249 self[0] = binsha
251 def _binsha(self):
252 return self[0]
254 binsha = property(_binsha, _set_binsha)
256 def _type(self):
257 return self[1]
259 def _set_type(self, type):
260 self[1] = type
262 type = property(_type, _set_type)
264 def _size(self):
265 return self[2]
267 def _set_size(self, size):
268 self[2] = size
270 size = property(_size, _set_size)
272 def _stream(self):
273 return self[3]
275 def _set_stream(self, stream):
276 self[3] = stream
278 stream = property(_stream, _set_stream)
280 #} END odb info interface
283class InvalidOInfo(tuple):
285 """Carries information about a sha identifying an object which is invalid in
286 the queried database. The exception attribute provides more information about
287 the cause of the issue"""
288 __slots__ = tuple()
290 def __new__(cls, sha, exc):
291 return tuple.__new__(cls, (sha, exc))
293 def __init__(self, sha, exc):
294 tuple.__init__(self, (sha, exc))
296 @property
297 def binsha(self):
298 return self[0]
300 @property
301 def hexsha(self):
302 return bin_to_hex(self[0])
304 @property
305 def error(self):
306 """:return: exception instance explaining the failure"""
307 return self[1]
310class InvalidOStream(InvalidOInfo):
312 """Carries information about an invalid ODB stream"""
313 __slots__ = tuple()
315#} END ODB Bases