Page MenuHomeSoftware Heritage
Paste P331

(An Untitled Masterwork)
ActivePublic

Authored by vlorentz on Nov 9 2018, 1:07 PM.
def _naive_sig(param_names):
return inspect.Signature([
inspect.Parameter(name, inspect.Parameter.POSITIONAL_OR_KEYWORD)
for name in param_names])
class Table:
def __init__(self, name, keys, columns):
if not set(keys).issubset(set(columns)):
raise ValueError('%r is not a subset of %r' %
(keys, columns))
if 'default' in keys:
raise ValueError('"default" is not a valid key name.')
self._keys = keys
self._columns = columns
self._sig_add = _naive_sig(columns)
self._sig_get = _naive_sig(keys)
self.Keys = collections.namedtuple(name + 'Keys', keys)
self.Record = collections.namedtuple(name + 'Record', columns)
self._data = {}
def _make_key(self, record):
return self.Keys(*(getattr(record, col) for col in self._keys))
def _make_record(self, args, kwargs):
bound_args = self._sig_add.bind(*args, **kwargs)
return self.Record(**bound_args.arguments)
def add(self, *args, **kwargs):
record = self._make_record(args, kwargs)
key = self._make_key(record)
if key in self._data:
raise DuplicateKey(key)
self._data[key] = record
def replace(self, *args, **kwargs):
record = self._make_record(args, kwargs)
key = self._make_key(record)
self._data[key] = record
def contains(self, *args, **kwargs):
bound_args = self._sig_get.bind(*args, **kwargs)
key = self.Keys(**bound_args.arguments)
return key in self._data
def get(self, *args, default=_TABLE_GET_NO_DEFAULT, **kwargs):
bound_args = self._sig_get.bind(*args, **kwargs)
key = self.Keys(**bound_args.arguments)
if default is _TABLE_GET_NO_DEFAULT:
return self._data[key]
else:
return self._data.get(key, default)
def get_dict(self, *args, **kwargs):
val = self.get(*args, **kwargs)
if val is None:
return None
return dict(zip(self._columns, map(copy.deepcopy, val)))
class Storage:
def __init__(self):
# ...
self._directories = Table('Directories', ['id'], ['id', 'entries'])
self._revisions = Table('Revisions', ['id'],
['id', 'message', 'author', 'date',
'committer', 'committer_date', 'parents',
'type', 'directory', 'metadata',
'synthetic'])
self._releases = Table('Releases', ['id'],
['id', 'name', 'author', 'date', 'target',
'target_type', 'message', 'synthetic'])
self._snapshots = Table('Snapshots', ['id'],
['id', 'origin', 'visit', 'branches',
'sorted_branch_names'])
self._origins = Table('Origins', ['type', 'url'],
['type', 'url'])