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'])